macrobase.py skeleton

core/src/python/.gitignore

@@ -0,0 +1 @@
+*.pyc

core/src/python/macrobase/__init__.py

@@ -0,0 +1,2 @@
+
+from macrobase import Pipeline

core/src/python/macrobase/distribution.py

@@ -0,0 +1,14 @@
+
+import macrobase as mb
+import pandas as pd
+
+class BatchMAD(mb.Classifier):
+    ''' Batch classifier computes Median Absolute Deviation
+        and compares to specified threshold '''
+    def __init__(self, threshold = 3):
+        self.threshold = threshold
+
+    def process(self, batch):
+        self.median = batch.median()
+        self.mad = (abs(batch - self.median)).median()
+        return pd.DataFrame(abs(batch-self.median)/self.mad > self.threshold)

core/src/python/macrobase/feature.py

@@ -0,0 +1,9 @@
+
+import macrobase as mb
+
+class LambdaTransform(mb.FeatureTransform):
+    def __init__(self, func):
+        self.func = func
+
+    def process(self, batch):
+        return batch.apply(self.func)

core/src/python/macrobase/macrobase.py

@@ -0,0 +1,42 @@
+
+import abc
+
+import pandas as pd
+
+class Pipeline(object):
+    """Core MacroBase execution unit.
+       Currently implements non-branching dataflow."""
+    def __init__(self):
+        self.operators = []
+
+    def then(self, operator):
+        self.operators.append(operator)
+        return self
+
+    def add(self, operator):
+        return self.then(operator)
+
+    def process(self, data):
+        for op in self.operators:
+            data = op.process(data)
+        return data
+
+class Operator(object):
+    """MacroBase dataflow operator"""    
+    def __init__(self):
+        pass
+
+    @abc.abstractmethod
+    def process(self, batch):
+        raise NotImplementedError("Calling an abstract method.")
+
+class FeatureTransform(Operator):
+    pass
+
+class Classifier(Operator):
+    pass
+
+
+
+
+

core/src/python/sample/sample.py

@@ -0,0 +1,16 @@
+import os
+import sys
+sys.path.insert(0, os.path.abspath('..'))
+
+import pandas as pd
+
+from macrobase import Pipeline
+from macrobase.distribution import BatchMAD
+from macrobase.feature import LambdaTransform
+
+p = Pipeline()
+data = pd.DataFrame([1,2,4,3,123,124,12,5109120398,12923012,123,123])
+
+result = p.add(LambdaTransform(lambda x: x+2)).then(BatchMAD()).process(data)
+print result
+

core/src/python/test/test_distribution.py

@@ -0,0 +1,21 @@
+import os
+import sys
+sys.path.insert(0, os.path.abspath('..'))
+
+import unittest
+from pandas.util.testing import assert_frame_equal
+
+from macrobase import Pipeline
+from macrobase.distribution import BatchMAD
+
+import pandas as pd
+import numpy as np
+
+class TestMAD(unittest.TestCase):
+    def test(self):
+        s = pd.Series([1.5, 50, 2, 3, 10000])
+
+        m = BatchMAD()
+        result = m.process(s)
+        assert_frame_equal(result,
+                           pd.DataFrame([False, True, False, False, True]))