Basic version of create and predict segment
This commit is contained in:
Stuart Lynn
@@ -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;
|
||||
|
||||
@@ -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()))
|
||||
|
||||
Reference in New Issue
Block a user