diff --git a/src/pg/sql/05_segmentation.sql b/src/pg/sql/05_segmentation.sql index 00ced3f..7def90d 100644 --- a/src/pg/sql/05_segmentation.sql +++ b/src/pg/sql/05_segmentation.sql @@ -1,55 +1,12 @@ -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' + variable_name TEXT, + target_table TEXT ) -RETURNS TABLE (the_geom geometry, geoid text, prediction Numeric ) +RETURNS TABLE (cartodb_id text, prediction Numeric ) AS $$ - from crankshaft import segmentation + from crankshaft.segmentation import create_and_predict_segment # TODO: use named parameters or a dictionary - return segmentation.create_and_predict_segment(segment_name,query,geoid_column,census_table,target_table,'random_forest') + return create_and_predict_segment(query,variable_name,target_table) $$ LANGUAGE plpythonu; diff --git a/src/py/crankshaft/crankshaft/segmentation/segmentation.py b/src/py/crankshaft/crankshaft/segmentation/segmentation.py index 0a1a8da..3881733 100644 --- a/src/py/crankshaft/crankshaft/segmentation/segmentation.py +++ b/src/py/crankshaft/crankshaft/segmentation/segmentation.py @@ -4,193 +4,90 @@ 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.ensemble import GradientBoostingClassifier 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): +def replace_nan_with_mean(array): + indices = np.where(np.isnan(array)) #returns an array of rows and column indices + for row, col in zip(*indices): + array[row,col] = np.mean(array[~np.isnan(array[:,col]), col]) + return array + +def get_data(variable, feature_columns, query): + columns = ','.join(['array_agg("{col}") as "{col}"'.format(col=col) for col in feature_columns]) + data = plpy.execute(''' select array_agg("{variable}") as target, {columns} from ({query}) as a'''.format( + variable = variable, + columns = columns, + query = query + )) + target = np.array(data[0]['target']) + features = np.column_stack([np.array(data[0][col], dtype=float) for col in feature_columns]) + return replace_nan_with_mean(target), replace_nan_with_mean(features) + +def create_and_predict_segment(query,variable,target_query): """ 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]) + columns = plpy.execute('select * from ({query}) a limit 1 '.format(query=query))[0].keys() + + feature_columns = set(columns) - set([variable, 'the_geom', 'the_geom_webmercator']) + target,features = get_data(variable, feature_columns, query) + 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 + cartodb_ids, result = predict_segment(model,feature_columns,target_query) + return zip(cartodb_ids, result) -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)) + model = GradientBoostingClassifier(n_estimators = 200, max_features=features.shape[1]) plpy.notice('training the model: fitting to data') model.fit(features_train, target_train) - plpy.notice('training the model: fitting one') + plpy.notice('model trained') accuracy = calculate_model_accuracy(model,features,target) - return model, accuracy, features.columns + return model, accuracy 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): +def predict_segment(model,features,target_query): """ 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())) + batch_size = 1000 + joined_features = ','.join(['"{0}"::numeric'.format(a) for a in features]) + + cursor = plpy.cursor('select Array[{joined_features}] features from ({target_query}) a'.format( + joined_features=joined_features, + target_query= target_query + )) + results = [] + while True: rows = cursor.fetch(batch_size) if not rows: break + batch = np.row_stack([np.array(row['features'], dtype=float) for row in rows]) - batch = pd.DataFrame(query_to_dictionary(rows)) - batch_features = batch.dropna(axis =1, how='all').fillna(0) - prediction = model.predict(batch_features) + #Need to fix this. Should be global mean. This will cause weird effects + batch = replace_nan_with_mean(batch) + plpy.notice(len(batch)) + prediction = model.predict(batch) results.append(prediction) plpy.notice('predicting: predicted') - return [a['the_geom'] for a in geoms], [a['geoid'] for a in geo_ids],prediction + + cartodb_ids = plpy.execute('select array_agg(cartodb_id order by cartodb_id) as cartodb_ids from ({0}) a '.format(target_query))[0]['cartodb_ids'] + return cartodb_ids, np.concatenate(results) -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()))