Implementation of http://stackoverflow.com/questions/22999487/update-the-average-of-a-continuous-sequence-of-numbers-in-constant-time# in Python

test_avg.py

#!/usr/bin/env python


from functools import total_ordering


@total_ordering
class avg(object):
	@property
	def empty(self):
		return self.length <= 0
	@property
	def value(self):
		return self.sum / float(self.length)
	def __float__(self):
		return float(self.value)
	def __len__(self):
		return self.length
	def __add__(self, other):
		return type(self)(self).add(other)
	def __sub__(self, other):
		return type(self)(self).subtract(other)
	def __eq__(self, other):
		return (self is other or 
			(isinstance(other, avg) and self.value == other.value) or
			(self.value == other)
		)
	def __le__(self, other):
		return self.value < other.value
	def __str__(self):
		return str(self.value)


class value_avg(avg):
	def __init__(self, other = None):
		self._average = 0
		self._length = 0
		if other is not None:
			self._average = other._average
			self._length = other._length
	@staticmethod
	def new_with(*values):
		return value_avg().add(*values)
	@property
	def sum(self):
		return self._average * self._length
	@property
	def value(self):
		return self._average
	@property
	def length(self):
		return self._length
	def add(self, *values):
		ret = type(self)(self)
		for value in values:
			if isinstance(value, avg):
				ret._average = (ret.sum + value.sum) / (ret.length + value.length)
				ret._length = ret.length + value.length
			else:
				ret._average = (ret.sum + value) / (ret._length + 1.0)
				ret._length = ret._length + 1
		return ret
	def subtract(self, *values):
		ret = type(self)(self)
		for value in values:
			if isinstance(value, avg):
				ret._average = (ret.sum - value.sum) / (ret.length - value.length)
				ret._length = ret.length - value.length
			else:
				ret._average = (ret.sum - value) / (ret.length - 1.0)
				ret._length = ret.length - 1
		return ret


class list_avg(avg):
	def __init__(self, other = None):
		self._values = []
		if other is not None:
			self._values = [x for x in other._values]
	@staticmethod
	def new_with(*values):
		return list_avg().add(*values)
	@property
	def sum(self):
		ret = 0
		for value in self._values:
			if isinstance(value, avg):
				ret += value.sum
			else:
				ret += value
		return ret
	@property
	def length(self):
		ret = 0
		for value in self._values:
			if isinstance(value, avg):
				ret += value.length
			else:
				ret += 1
		return ret
	def add(self, *values):
		ret = type(self)(self)
		for value in values: ret._values.append(value)
		return ret
	def subtract(self, *values):
		ret = type(self)(self)
		for value in values:
			if value in ret._values: ret._values.remove(value)
		return ret


def standard_avg(*values):
	return reduce(lambda a, b: a + b, values) / len(values)


values = [1, 5, 2, 10, 3.5]
savg = standard_avg(*values)
vavg = value_avg.new_with(*values)
lavg = list_avg.new_with(*values)
print("[values] standard avg.: %f, value-based avg.: %f, list-based avg.: %f" % (savg, vavg, lavg))
values.remove(5)
savg = standard_avg(*values)
vavg = vavg - 5
lavg = lavg - 5
print("[values - 5] standard avg.: %f, value-based avg.: %f, list-based avg.: %f" % (savg, vavg, lavg))
vavg, lavg = vavg + lavg, lavg + vavg
print("[lavg + vavg] value-based avg.: %f, list-based avg.: %f" % (vavg, lavg))
# What am I even doing?
ravg = value_avg.new_with(5, 2, 10, 3.5)
vavg, lavg = vavg + ravg, lavg + ravg
tavg = standard_avg(*([1, 2, 10, 3.5] + [5, 2, 10, 3.5]))
print("[lavg/vavg + ravg] new avg.: %f, real avg.: %f, value-based avg.: %f, list-based avg.: %f" % (ravg, tavg, vavg, lavg))