changing to new crankshaft format

src/pg/sql/05_segmentation.sql

@@ -0,0 +1,55 @@
+CREATE OR REPLACE FUNCTION
+  CDB_CreateSegment (
+      segment_name TEXT,
+      table_name TEXT,
+          column_name TEXT,
+      geoid_column TEXT DEFAULT 'geoid',
+      census_table TEXT DEFAULT 'block_groups'
+  )
+RETURNS NUMERIC
+AS $$
+  from crankshaft import segmentation
+  # TODO: use named parameters or a dictionary
+  return segmentation.create_segment(segment_name,table_name,column_name,geoid_column,census_table,'random_forest')
+$$ LANGUAGE plpythonu;
+
+CREATE OR REPLACE FUNCTION
+  CDB_CorrelatedVariables(
+    query text,
+    geoid_column text DEFAULT 'geoid',
+    census_table text DEFAULT 'ml_learning_block_groups_clipped'
+  )
+RETURNS TABLE(feature text, importance NUMERIC, std NUMERIC)
+AS $$
+  from crankshaft.segmentation import correlated_variables
+  return correlated_variables(query,geoid_column,census_table)
+$$ LANGUAGE plpythonu;
+
+CREATE OR REPLACE FUNCTION
+  CDB_PredictSegment (
+      segment_name TEXT,
+      geoid_column TEXT DEFAULT 'geoid',
+      census_table TEXT DEFAULT 'block_groups'
+  )
+RETURNS TABLE(geoid TEXT, prediction NUMERIC)
+AS $$
+  from crankshaft.segmentation import create_segemnt
+  # TODO: use named parameters or a dictionary
+  return create_segment('table')
+$$ LANGUAGE plpythonu;
+
+
+CREATE OR REPLACE FUNCTION
+  CDB_CreateAndPredictSegment (
+      segment_name TEXT,
+      query TEXT,
+      target_table TEXT,
+      geoid_column TEXT DEFAULT 'geoid',
+      census_table TEXT DEFAULT 'block_groups'
+  )
+RETURNS TABLE (the_geom geometry, geoid text, prediction Numeric )
+AS $$
+  from crankshaft import segmentation
+  # TODO: use named parameters or a dictionary
+  return segmentation.create_and_predict_segment(segment_name,query,geoid_column,census_table,target_table,'random_forest')
+$$ LANGUAGE plpythonu;

src/py/crankshaft/crankshaft/__init__.py

@@ -1,2 +1,3 @@
 import random_seeds
 import clustering
+import segmentation

src/py/crankshaft/crankshaft/segmentation/__init__.py

@@ -0,0 +1 @@
+from segmentation import * 

src/py/crankshaft/crankshaft/segmentation/segmentation.py

@@ -0,0 +1,196 @@
+"""
+Segmentation creation and prediction
+"""
+
+import sklearn
+import numpy as np
+import pandas as pd
+import cPickle
+import plpy
+import sys
+from sklearn.ensemble import ExtraTreesRegressor
+from sklearn import metrics
+from sklearn.externals import joblib
+from sklearn.cross_validation import train_test_split
+import StringIO
+import gzip
+
+# High level interface ---------------------------------------
+
+def create_segment(segment_name,table_name,column_name,geoid_column,census_table,method):
+    """
+    generate a segment with machine learning
+    Stuart Lynn
+    """
+    data     = pd.DataFrame(join_with_census(table_name, column_name,geoid_column, census_table))
+    features = data[data.columns.difference([column_name, 'geoid','the_geom', 'the_geom_webmercator'])]
+    target, mean, std = normalize(data[column_name])
+    model, accuracy = train_model(target,features, test_split=0.2)
+    save_model(segment_name, model, accuracy, table_name, column_name, census_table, geoid_column, method)
+    # predict_segment
+    return accuracy
+
+def correlated_variables(query,geoid_column,census_table):
+    """
+    returns the columns which are importaint for the random forrest model
+    """
+    data     = pd.DataFrame(join_with_census(query,geoid_column, census_table))
+    features = data[data.columns.difference(['target', 'the_geom_webmercator', 'geoid','the_geom'])]
+    target, mean, std = normalize(data['target'])
+    model, accuracy, used_features = train_model(target,features, test_split=0.2)
+    std = np.std([tree.feature_importances_ for tree in model.estimators_],
+             axis=0)
+    importances = model.feature_importances_
+    return zip(features,importances,std)
+
+
+def create_and_predict_segment(segment_name,query,geoid_column,census_table,target_table,method):
+    """
+    generate a segment with machine learning
+    Stuart Lynn
+    """
+    data     = pd.DataFrame(join_with_census(query,geoid_column, census_table))
+    features = data[data.columns.difference(['target', 'the_geom_webmercator', 'geoid','the_geom'])]
+    target, mean, std = normalize(data['target'])
+
+    normed_target,target_mean, target_std = normalize(target)
+    plpy.notice('mean ', target_mean, " std ", target_std)
+    model, accuracy, used_features = train_model(target,features, test_split=0.2)
+    # save_model(segment_name, model, accuracy, table_name, column_name, census_table, geoid_column, method)
+    geoms, geoids, result = predict_segment(model,used_features,geoid_column,target_table)
+    return zip(geoms,geoids, [denormalize(t,target_mean, target_std) for t in result] )
+
+
+def normalize(target):
+    mean = np.mean(target)
+    std  = np.std(target)
+    plpy.notice('mean '+str(mean)+" std : "+str(std))
+    return (target - mean)/std, mean, std
+
+def denormalize(target, mean ,std):
+    return target*std + mean
+
+def train_model(target,features,test_split):
+    plpy.notice('training the model')
+    plpy.notice('before ', str(np.shape(features)))
+    features = features.dropna(axis =1, how='all').fillna(0)
+    plpy.notice('after ', str(np.shape(features)))
+    target = target.fillna(0)
+    features_train, features_test, target_train, target_test = train_test_split(features, target, test_size=test_split)
+    model = ExtraTreesRegressor(n_estimators = 200, max_features=len(features.columns))
+    plpy.notice('training the model: fitting to data')
+    model.fit(features_train, target_train)
+    plpy.notice('training the model: fitting one')
+    accuracy = calculate_model_accuracy(model,features,target)
+    return model, accuracy, features.columns
+
+def calculate_model_accuracy(model,features,target):
+    prediction = model.predict(features)
+    return metrics.mean_squared_error(prediction,target)/np.std(target)
+
+def join_with_census(query, geoid_column, census_table):
+    columns          = plpy.execute('select * from  {census_table}  limit 1  '.format(**locals()))
+    combined_columns = [ a for a in columns[0].keys() if a not in ['target','the_geom','cartodb_id','geoid','the_geom_webmercator']]
+    plpy.notice(combined_columns)
+    feature_names    = ",".join([ " {census_table}.\"{a}\"::Numeric as \"{a}\" ".format(**locals()) for a in combined_columns])
+    plpy.notice(feature_names)
+
+    plpy.notice('joining with census data')
+    join_data     = plpy.execute('''
+
+        SELECT {feature_names}, a.target
+        FROM   ({query}) a
+        JOIN   {census_table}
+        ON  a.{geoid_column}::numeric = {census_table}.geoid::numeric
+    '''.format(**locals()))
+
+    if len(join_data) == 0:
+        plpy.notice('Failed to join with census data')
+
+    return  query_to_dictionary(join_data)
+
+def query_to_dictionary(result):
+    return [ dict(zip(r.keys(), r.values())) for r in result ]
+
+def predict_segment(model,features,geoid_column,census_table):
+    """
+    predict a segment with machine learning
+    Stuart Lynn
+    """
+    # data     = fetch_model(segment_name)
+    # model    = data['model']
+    # features = ",".join(features)
+
+    joined_features  = ','.join(['\"'+a+'\"::numeric' for a in features])
+      = plpy.execute()
+    cursor = plpy.cursor('select {joined_features} from {census_table}'.format(**locals()))
+    results = []
+    while True:
+        rows  = cursor.fetch(batch_size)
+
+        if not rows:
+            break
+
+        batch = pd.DataFrame(query_to_dictionary(rows))
+        batch_features = batch.dropna(axis =1, how='all').fillna(0)
+        prediction   = model.predict(batch_features)
+        results.append(prediction)
+        plpy.notice('predicting: predicted')
+
+    return  [a['the_geom'] for a in geoms], [a['geoid'] for a in geo_ids],prediction
+
+
+def fetch_model(model_name):
+    """
+    fetch a model from storage
+    """
+    data = plpy.execute('select * from models where name={model_name}')
+    if len(data)==0:
+        plpy.notice('model not found')
+    data = data[0]
+    data['model'] = pickle.load(data['model'])
+    return data
+
+
+def create_model_table():
+    """
+    create the model table if requred
+    """
+    plpy.execute('''
+        CREATE table IF NOT EXISTS _cdb_models(
+            name TEXT,
+            model TEXT,
+            features TEXT[],
+            accuracy NUMERIC,
+            table_name TEXT,
+            census_table_name TEXT,
+            method TEXT
+    )''')
+
+def save_model(model_name,model,accuracy,table_name, column_name,census_table,geoid_column,method):
+    """
+    save a model to the model table for later use
+    """
+    create_model_table()
+
+    plpy.execute('''
+        DELETE FROM _cdb_models WHERE name = '{model_name}'
+    '''.format(**locals()))
+
+    # stringio = StringIO.StringIO()
+    # gzip_file = gzip.GzipFile(fileobj=stringio, mode='w')
+    # gzip_file.write()
+    # gzip_file.close()
+
+    model_pickle = cPickle.dumps(model) #stringio.getvalue()
+
+
+    # stringio.close()
+
+    plpy.notice(type(model_pickle))
+    plpy.notice(len(model_pickle))
+    plpy.notice(sys.getsizeof(model_pickle))
+    model_pickle  =plpy.quote_literal(model_pickle)
+    plpy.execute("""
+        INSERT INTO _cdb_models VALUES ('{model_name}',$${model_pickle}$$, Array['test1', 'test2'],{accuracy}, '{table_name}', '{census_table}', '{method}')
+    """.format(**locals()))