Testing two approaches to subsetting a long (MTS) list

trim_performance_test.py

import datetime
import logging
import random
import time

import numpy as np

from tscached.mts import MTS
from testing.mock_redis import MockRedis

"""
Investigate which trim (subset) algorithms are more performant.
Relies on numpy, plus github.com/zachm/tscached

Example results:
(venv) ztm@140-sfoeng51-244:~/sauce/tscached$ time python trim_performance_test.py
Efficient, one slice, n=100, units=ms
        mean  :  0.0395512580872
        median:  0.0338554382324
        stddev:  0.0124302262613
Efficient, two slices, n=100, units=ms
        mean  :  0.0389266014099
        median:  0.0319480895996
        stddev:  0.0127830097245
Robust, one slice, n=100, units=ms
        mean  :  1.13894224167
        median:  1.06203556061
        stddev:  0.260296729155
Robust, two slices, n=100, units=ms
        mean  :  1.43874168396
        median:  1.37758255005
        stddev:  0.179611378103

real    0m6.492s
user    0m6.386s
sys     0m0.087s
"""


logging.getLogger().setLevel(logging.ERROR)


def robust_trim(data, start, end=None):
    """ This is a silly trim algorithm. Full O(n), but very robust.
        data: list of 2-ary lists: [timestamp in ms, value]
        start: datetime of range start
        end: datetime of range end, or None if returning until NOW.
        Returns: generator of 2-ary lists that match start, end constraints.
    """
    start = int(start.strftime('%s'))
    if end:
        end = int(end.strftime('%s'))
    for entry in data:
        if (entry[0] / 1000) >= start:
            if not end:
                yield entry
            elif (entry[0] / 1000) <= end:
                yield entry


def make_data():
    """ Avoid any sneaky compiler tricks by remaking your data set. """
    data = []
    for i in xrange(10000):
        data.append([(1234567890 + i) * 1000, random.randint(0, 5000)])
    return data


def test_efficient_trim(start_dt, end_dt, verbose=False):
    tester = MTS(MockRedis())
    tester.result = {'values': make_data()}

    start = time.time()
    result = tester.trim(start_dt, end_dt)
    elapsed = (time.time() - start) * 1000
    if verbose:
        print "existing trim took this long: (ms) ", elapsed
        print "values returned: %d, first %d, last %d" % (len(result),
                result[0][0], result[-1][0])
    return elapsed


def test_robust_trim(start_dt, end_dt, verbose=False):
    data = make_data()
    start = time.time()
    result = list(robust_trim(data, start_dt, end_dt))
    elapsed = (time.time() - start) * 1000
    if verbose:
        print "stupid  trim took this long: (ms) ", elapsed
        print "values returned: %d, first %d, last %d" % (len(result),
                result[0][0], result[-1][0])
    return elapsed


def summary_stats(lst):
    a = np.array(lst)
    print "\tmean  : ", np.mean(a)
    print "\tmedian: ", np.median(a)
    print "\tstddev: ", np.std(a)


efficients_single = []
efficients_double = []
robusts_single = []
robusts_double = []

start_dt = datetime.datetime.fromtimestamp(1234567890 + 5000)
end_dt = datetime.datetime.fromtimestamp(1234567890 + 7500)

for _ in xrange(100):
    efficients_single.append(test_efficient_trim(start_dt, None))
print "Efficient, one slice, n=100, units=ms"
summary_stats(efficients_single)

for _ in xrange(100):
    efficients_double.append(test_efficient_trim(start_dt, end_dt))
print "Efficient, two slices, n=100, units=ms"
summary_stats(efficients_double)

for _ in xrange(100):
    robusts_single.append(test_robust_trim(start_dt, None))
print "Robust, one slice, n=100, units=ms"
summary_stats(robusts_single)

for _ in xrange(100):
    robusts_double.append(test_robust_trim(start_dt, end_dt))
print "Robust, two slices, n=100, units=ms"
summary_stats(robusts_double)