<s> """<STR_LIT>""" <EOL> from __future__ import print_function <EOL> import logging <EOL> import numpy as np <EOL> import pandas as pd <EOL> import statsmodels . formula . api as smf <EOL> from patsy import dmatrix <EOL> from prettytable import PrettyTable <EOL> from zbox import toolz as tz <EOL> from . import util <EOL> from . . exceptions import ModelEvaluationError <EOL> from . . utils import yamlio <EOL> from . . utils . logutil import log_start_finish <EOL> logger = logging . getLogger ( __name__ ) <EOL> def fit_model ( df , filters , model_expression ) : <EOL> """<STR_LIT>""" <EOL> df = util . apply_filter_query ( df , filters ) <EOL> model = smf . ols ( formula = model_expression , data = df ) <EOL> if len ( model . exog ) != len ( df ) : <EOL> raise ModelEvaluationError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> with log_start_finish ( '<STR_LIT>' , logger ) : <EOL> return model . fit ( ) <EOL> def predict ( df , filters , model_fit , ytransform = None ) : <EOL> """<STR_LIT>""" <EOL> df = util . apply_filter_query ( df , filters ) <EOL> with log_start_finish ( '<STR_LIT>' , logger ) : <EOL> sim_data = model_fit . predict ( df ) <EOL> if len ( sim_data ) != len ( df ) : <EOL> raise ModelEvaluationError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if ytransform : <EOL> sim_data = ytransform ( sim_data ) <EOL> return pd . Series ( sim_data , index = df . index ) <EOL> def _rhs ( model_expression ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT>' not in model_expression : <EOL> return model_expression <EOL> else : <EOL> return model_expression . split ( '<STR_LIT>' ) [ <NUM_LIT:1> ] . strip ( ) <EOL> class _FakeRegressionResults ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , model_expression , fit_parameters , rsquared , <EOL> rsquared_adj ) : <EOL> self . model_expression = model_expression <EOL> self . params = fit_parameters [ '<STR_LIT>' ] <EOL> self . bse = fit_parameters [ '<STR_LIT>' ] <EOL> self . tvalues = fit_parameters [ '<STR_LIT>' ] <EOL> self . rsquared = rsquared <EOL> self . rsquared_adj = rsquared_adj <EOL> @ property <EOL> def _rhs ( self ) : <EOL> """<STR_LIT>""" <EOL> return _rhs ( self . model_expression ) <EOL> def predict ( self , data ) : <EOL> """<STR_LIT>""" <EOL> with log_start_finish ( '<STR_LIT>' , logger ) : <EOL> model_design = dmatrix ( <EOL> self . _rhs , data = data , return_type = '<STR_LIT>' ) <EOL> return model_design . dot ( self . params ) . values <EOL> def _model_fit_to_table ( fit ) : <EOL> """<STR_LIT>""" <EOL> fit_parameters = pd . DataFrame ( <EOL> { '<STR_LIT>' : fit . params , <EOL> '<STR_LIT>' : fit . bse , <EOL> '<STR_LIT>' : fit . tvalues } ) <EOL> fit_parameters . rsquared = fit . rsquared <EOL> fit_parameters . rsquared_adj = fit . rsquared_adj <EOL> return fit_parameters <EOL> YTRANSFORM_MAPPING = { <EOL> None : None , <EOL> np . exp : '<STR_LIT>' , <EOL> '<STR_LIT>' : np . exp , <EOL> np . log : '<STR_LIT>' , <EOL> '<STR_LIT>' : np . log , <EOL> np . log1p : '<STR_LIT>' , <EOL> '<STR_LIT>' : np . log1p , <EOL> np . expm1 : '<STR_LIT>' , <EOL> '<STR_LIT>' : np . expm1 <EOL> } <EOL> class RegressionModel ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , fit_filters , predict_filters , model_expression , <EOL> ytransform = None , name = None ) : <EOL> self . fit_filters = fit_filters <EOL> self . predict_filters = predict_filters <EOL> self . model_expression = model_expression <EOL> self . ytransform = ytransform <EOL> self . name = name or '<STR_LIT>' <EOL> self . model_fit = None <EOL> self . fit_parameters = None <EOL> self . est_data = None <EOL> @ classmethod <EOL> def from_yaml ( cls , yaml_str = None , str_or_buffer = None ) : <EOL> """<STR_LIT>""" <EOL> cfg = yamlio . yaml_to_dict ( yaml_str , str_or_buffer ) <EOL> model = cls ( <EOL> cfg [ '<STR_LIT>' ] , <EOL> cfg [ '<STR_LIT>' ] , <EOL> cfg [ '<STR_LIT>' ] , <EOL> YTRANSFORM_MAPPING [ cfg [ '<STR_LIT>' ] ] , <EOL> cfg [ '<STR_LIT:name>' ] ) <EOL> if '<STR_LIT>' in cfg and cfg [ '<STR_LIT>' ] : <EOL> fit_parameters = pd . DataFrame ( cfg [ '<STR_LIT>' ] ) <EOL> fit_parameters . rsquared = cfg [ '<STR_LIT>' ] <EOL> fit_parameters . rsquared_adj = cfg [ '<STR_LIT>' ] <EOL> model . model_fit = _FakeRegressionResults ( <EOL> model . str_model_expression , <EOL> fit_parameters , <EOL> cfg [ '<STR_LIT>' ] , cfg [ '<STR_LIT>' ] ) <EOL> model . fit_parameters = fit_parameters <EOL> logger . debug ( '<STR_LIT>' . format ( model . name ) ) <EOL> return model <EOL> @ property <EOL> def str_model_expression ( self ) : <EOL> """<STR_LIT>""" <EOL> return util . str_model_expression ( <EOL> self . model_expression , add_constant = True ) <EOL> def fit ( self , data , debug = False ) : <EOL> """<STR_LIT>""" <EOL> with log_start_finish ( '<STR_LIT>' . format ( self . name ) , logger ) : <EOL> fit = fit_model ( data , self . fit_filters , self . str_model_expression ) <EOL> self . model_fit = fit <EOL> self . fit_parameters = _model_fit_to_table ( fit ) <EOL> if debug : <EOL> index = util . apply_filter_query ( data , self . fit_filters ) . index <EOL> assert len ( fit . model . exog ) == len ( index ) , ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> df = pd . DataFrame ( <EOL> fit . model . exog , columns = fit . model . exog_names , index = index ) <EOL> df [ fit . model . endog_names ] = fit . model . endog <EOL> df [ "<STR_LIT>" ] = fit . fittedvalues <EOL> df [ "<STR_LIT>" ] = fit . resid <EOL> self . est_data = df <EOL> return fit <EOL> @ property <EOL> def fitted ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . model_fit is not None <EOL> def assert_fitted ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . fitted : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> def report_fit ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . fitted : <EOL> print ( '<STR_LIT>' ) <EOL> return <EOL> print ( '<STR_LIT>' . format ( self . model_fit . rsquared ) ) <EOL> print ( '<STR_LIT>' . format ( self . model_fit . rsquared_adj ) ) <EOL> print ( '<STR_LIT>' ) <EOL> tbl = PrettyTable ( <EOL> [ '<STR_LIT>' , ] ) <EOL> tbl = PrettyTable ( ) <EOL> tbl . add_column ( '<STR_LIT>' , self . fit_parameters . index . values ) <EOL> for col in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> tbl . add_column ( col , self . fit_parameters [ col ] . values ) <EOL> tbl . align [ '<STR_LIT>' ] = '<STR_LIT:l>' <EOL> tbl . float_format = '<STR_LIT>' <EOL> print ( tbl ) <EOL> def predict ( self , data ) : <EOL> """<STR_LIT>""" <EOL> self . assert_fitted ( ) <EOL> with log_start_finish ( '<STR_LIT>' . format ( self . name ) , logger ) : <EOL> return predict ( <EOL> data , self . predict_filters , self . model_fit , self . ytransform ) <EOL> def to_dict ( self ) : <EOL> """<STR_LIT>""" <EOL> d = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : self . name , <EOL> '<STR_LIT>' : self . fit_filters , <EOL> '<STR_LIT>' : self . predict_filters , <EOL> '<STR_LIT>' : self . model_expression , <EOL> '<STR_LIT>' : YTRANSFORM_MAPPING [ self . ytransform ] , <EOL> '<STR_LIT>' : self . fitted , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None <EOL> } <EOL> if self . fitted : <EOL> d [ '<STR_LIT>' ] = yamlio . frame_to_yaml_safe ( <EOL> self . fit_parameters ) <EOL> d [ '<STR_LIT>' ] = float ( self . model_fit . rsquared ) <EOL> d [ '<STR_LIT>' ] = float ( self . model_fit . rsquared_adj ) <EOL> return d <EOL> def to_yaml ( self , str_or_buffer = None ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( <EOL> '<STR_LIT>' . format ( self . name ) ) <EOL> return yamlio . convert_to_yaml ( self . to_dict ( ) , str_or_buffer ) <EOL> def columns_used ( self ) : <EOL> """<STR_LIT>""" <EOL> return list ( tz . unique ( tz . concatv ( <EOL> util . columns_in_filters ( self . fit_filters ) , <EOL> util . columns_in_filters ( self . predict_filters ) , <EOL> util . columns_in_formula ( self . model_expression ) ) ) ) <EOL> @ classmethod <EOL> def fit_from_cfg ( cls , df , cfgname , debug = False ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( '<STR_LIT>' . format ( cfgname ) ) <EOL> hm = cls . from_yaml ( str_or_buffer = cfgname ) <EOL> ret = hm . fit ( df , debug = debug ) <EOL> print ( ret . summary ( ) ) <EOL> hm . to_yaml ( str_or_buffer = cfgname ) <EOL> logger . debug ( '<STR_LIT>' . format ( cfgname ) ) <EOL> return hm <EOL> @ classmethod <EOL> def predict_from_cfg ( cls , df , cfgname ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( '<STR_LIT>' . format ( cfgname ) ) <EOL> hm = cls . from_yaml ( str_or_buffer = cfgname ) <EOL> price_or_rent = hm . predict ( df ) <EOL> print ( price_or_rent . describe ( ) ) <EOL> logger . debug ( '<STR_LIT>' . format ( cfgname ) ) <EOL> return price_or_rent , hm <EOL> class RegressionModelGroup ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , segmentation_col , name = None ) : <EOL> self . segmentation_col = segmentation_col <EOL> self . name = name if name is not None else '<STR_LIT>' <EOL> self . models = { } <EOL> def add_model ( self , model ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( <EOL> '<STR_LIT>' . format ( model . name , self . name ) ) <EOL> self . models [ model . name ] = model <EOL> def add_model_from_params ( self , name , fit_filters , predict_filters , <EOL> model_expression , ytransform = None ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( <EOL> '<STR_LIT>' . format ( name , self . name ) ) <EOL> model = RegressionModel ( <EOL> fit_filters , predict_filters , model_expression , ytransform , name ) <EOL> self . models [ name ] = model <EOL> def _iter_groups ( self , data ) : <EOL> """<STR_LIT>""" <EOL> groups = data . groupby ( self . segmentation_col ) <EOL> for name in self . models : <EOL> yield name , groups . get_group ( name ) <EOL> def fit ( self , data , debug = False ) : <EOL> """<STR_LIT>""" <EOL> with log_start_finish ( <EOL> '<STR_LIT>' . format ( self . name ) , logger ) : <EOL> return { name : self . models [ name ] . fit ( df , debug = debug ) <EOL> for name , df in self . _iter_groups ( data ) } <EOL> @ property <EOL> def fitted ( self ) : <EOL> """<STR_LIT>""" <EOL> return ( all ( m . fitted for m in self . models . values ( ) ) <EOL> if self . models else False ) <EOL> def predict ( self , data ) : <EOL> """<STR_LIT>""" <EOL> with log_start_finish ( <EOL> '<STR_LIT>' . format ( self . name ) , logger ) : <EOL> results = [ self . models [ name ] . predict ( df ) <EOL> for name , df in self . _iter_groups ( data ) ] <EOL> return pd . concat ( results ) <EOL> def columns_used ( self ) : <EOL> """<STR_LIT>""" <EOL> return list ( tz . unique ( tz . concat ( <EOL> m . columns_used ( ) for m in self . models . values ( ) ) ) ) <EOL> class SegmentedRegressionModel ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( <EOL> self , segmentation_col , fit_filters = None , predict_filters = None , <EOL> default_model_expr = None , default_ytransform = None , <EOL> min_segment_size = <NUM_LIT:0> , name = None ) : <EOL> self . segmentation_col = segmentation_col <EOL> self . _group = RegressionModelGroup ( segmentation_col ) <EOL> self . fit_filters = fit_filters <EOL> self . predict_filters = predict_filters <EOL> self . default_model_expr = default_model_expr <EOL> self . default_ytransform = default_ytransform <EOL> self . min_segment_size = min_segment_size <EOL> self . name = name if name is not None else '<STR_LIT>' <EOL> @ classmethod <EOL> def from_yaml ( cls , yaml_str = None , str_or_buffer = None ) : <EOL> """<STR_LIT>""" <EOL> cfg = yamlio . yaml_to_dict ( yaml_str , str_or_buffer ) <EOL> default_model_expr = cfg [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> default_ytransform = cfg [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> seg = cls ( <EOL> cfg [ '<STR_LIT>' ] , cfg [ '<STR_LIT>' ] , <EOL> cfg [ '<STR_LIT>' ] , default_model_expr , <EOL> YTRANSFORM_MAPPING [ default_ytransform ] , cfg [ '<STR_LIT>' ] , <EOL> cfg [ '<STR_LIT:name>' ] ) <EOL> if "<STR_LIT>" not in cfg : <EOL> cfg [ "<STR_LIT>" ] = { } <EOL> for name , m in cfg [ '<STR_LIT>' ] . items ( ) : <EOL> m [ '<STR_LIT>' ] = m . get ( <EOL> '<STR_LIT>' , default_model_expr ) <EOL> m [ '<STR_LIT>' ] = m . get ( '<STR_LIT>' , default_ytransform ) <EOL> m [ '<STR_LIT>' ] = None <EOL> m [ '<STR_LIT>' ] = None <EOL> reg = RegressionModel . from_yaml ( yamlio . convert_to_yaml ( m , None ) ) <EOL> seg . _group . add_model ( reg ) <EOL> logger . debug ( <EOL> '<STR_LIT>' . format ( seg . name ) ) <EOL> return seg <EOL> def add_segment ( self , name , model_expression = None , ytransform = '<STR_LIT:default>' ) : <EOL> """<STR_LIT>""" <EOL> if not model_expression : <EOL> if self . default_model_expr is None : <EOL> raise ValueError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> model_expression = self . default_model_expr <EOL> if ytransform == '<STR_LIT:default>' : <EOL> ytransform = self . default_ytransform <EOL> self . _group . add_model_from_params ( <EOL> name , None , None , model_expression , ytransform ) <EOL> logger . debug ( '<STR_LIT>' . format ( name , self . name ) ) <EOL> def fit ( self , data , debug = False ) : <EOL> """<STR_LIT>""" <EOL> data = util . apply_filter_query ( data , self . fit_filters ) <EOL> unique = data [ self . segmentation_col ] . unique ( ) <EOL> value_counts = data [ self . segmentation_col ] . value_counts ( ) <EOL> gone = set ( self . _group . models ) - set ( unique ) <EOL> for g in gone : <EOL> del self . _group . models [ g ] <EOL> for x in unique : <EOL> if x not in self . _group . models and value_counts [ x ] > self . min_segment_size : <EOL> self . add_segment ( x ) <EOL> with log_start_finish ( <EOL> '<STR_LIT>' . format ( self . name ) , <EOL> logger ) : <EOL> return self . _group . fit ( data , debug = debug ) <EOL> @ property <EOL> def fitted ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _group . fitted <EOL> def predict ( self , data ) : <EOL> """<STR_LIT>""" <EOL> with log_start_finish ( <EOL> '<STR_LIT>' . format ( self . name ) , <EOL> logger ) : <EOL> data = util . apply_filter_query ( data , self . predict_filters ) <EOL> return self . _group . predict ( data ) <EOL> def _process_model_dict ( self , d ) : <EOL> """<STR_LIT>""" <EOL> del d [ '<STR_LIT>' ] <EOL> del d [ '<STR_LIT>' ] <EOL> del d [ '<STR_LIT>' ] <EOL> if d [ '<STR_LIT>' ] == self . default_model_expr : <EOL> del d [ '<STR_LIT>' ] <EOL> if YTRANSFORM_MAPPING [ d [ '<STR_LIT>' ] ] == self . default_ytransform : <EOL> del d [ '<STR_LIT>' ] <EOL> d [ "<STR_LIT:name>" ] = yamlio . to_scalar_safe ( d [ "<STR_LIT:name>" ] ) <EOL> return d <EOL> def to_dict ( self ) : <EOL> """<STR_LIT>""" <EOL> return { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : self . name , <EOL> '<STR_LIT>' : self . segmentation_col , <EOL> '<STR_LIT>' : self . fit_filters , <EOL> '<STR_LIT>' : self . predict_filters , <EOL> '<STR_LIT>' : self . min_segment_size , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : self . default_model_expr , <EOL> '<STR_LIT>' : YTRANSFORM_MAPPING [ self . default_ytransform ] <EOL> } , <EOL> '<STR_LIT>' : self . fitted , <EOL> '<STR_LIT>' : { <EOL> yamlio . to_scalar_safe ( name ) : <EOL> self . _process_model_dict ( m . to_dict ( ) ) <EOL> for name , m in self . _group . models . items ( ) } <EOL> } <EOL> def to_yaml ( self , str_or_buffer = None ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( <EOL> '<STR_LIT>' . format ( <EOL> self . name ) ) <EOL> return yamlio . convert_to_yaml ( self . to_dict ( ) , str_or_buffer ) <EOL> def columns_used ( self ) : <EOL> """<STR_LIT>""" <EOL> return list ( tz . unique ( tz . concatv ( <EOL> util . columns_in_filters ( self . fit_filters ) , <EOL> util . columns_in_filters ( self . predict_filters ) , <EOL> util . columns_in_formula ( self . default_model_expr ) , <EOL> self . _group . columns_used ( ) , <EOL> [ self . segmentation_col ] ) ) ) <EOL> @ classmethod <EOL> def fit_from_cfg ( cls , df , cfgname , debug = False , min_segment_size = None ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( '<STR_LIT>' . format ( cfgname ) ) <EOL> hm = cls . from_yaml ( str_or_buffer = cfgname ) <EOL> if min_segment_size : <EOL> hm . min_segment_size = min_segment_size <EOL> for k , v in hm . fit ( df , debug = debug ) . items ( ) : <EOL> print ( "<STR_LIT>" % str ( k ) ) <EOL> print ( v . summary ( ) ) <EOL> hm . to_yaml ( str_or_buffer = cfgname ) <EOL> logger . debug ( '<STR_LIT>' . format ( cfgname ) ) <EOL> return hm <EOL> @ classmethod <EOL> def predict_from_cfg ( cls , df , cfgname , min_segment_size = None ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( '<STR_LIT>' . format ( cfgname ) ) <EOL> hm = cls . from_yaml ( str_or_buffer = cfgname ) <EOL> if min_segment_size : <EOL> hm . min_segment_size = min_segment_size <EOL> price_or_rent = hm . predict ( df ) <EOL> print ( price_or_rent . describe ( ) ) <EOL> logger . debug ( '<STR_LIT>' . format ( cfgname ) ) <EOL> return price_or_rent , hm </s>
