gavindiaz/drift
1
0
Fork 0
mirror of https://github.com/webclinic017/drift.git synced 2026-07-27 18:57:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(Training): use date indexes instead of integers, need this to prepare for Events (#185)

Browse Source
This commit is contained in:
Mark Aron Szulyovszky
2022-01-24 12:22:30 +01:00
committed by GitHub
parent e80fffdb65
commit e6e2317fe0
15 changed files with 50 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/test.yml
+8
View File
@@ -28,6 +28,14 @@ jobs:
shell: bash -l {0}
run: |
python run_portfolio_reporting.py
- name: Run inference
shell: bash -l {0}
run: |
python run_inference.py
- name: Delete models
shell: bash -l {0}
run: |
rm -rf ./output/models
- uses: actions/upload-artifact@v2
with:
name: artifacts
config/config.py
-5
View File
@@ -27,7 +27,6 @@ class RawConfig(BaseModel):
other_features: list[str]
exogenous_features: list[str]
no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced']
index_column: Literal['date', 'int']
primary_models: list[str]
meta_labeling_models: list[str]
@@ -56,7 +55,6 @@ class Config(BaseModel):
other_features: list[tuple[str, FeatureExtractor, list[int]]]
exogenous_features: list[tuple[str, FeatureExtractor, list[int]]]
no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced']
index_column: Literal['date', 'int']
primary_models: list[tuple[str, Model]]
meta_labeling_models: list[tuple[str, Model]]
@@ -92,7 +90,6 @@ def get_dev_config() -> RawConfig:
own_features = ['level_2', 'date_days'],
other_features = ['single_mom'],
exogenous_features = ['z_score'],
index_column= 'int',
no_of_classes= 'two',
primary_models = classification_models,
@@ -129,7 +126,6 @@ def get_default_ensemble_config() -> RawConfig:
own_features = ['level_2', 'date_days', 'lags_up_to_5'],
other_features = ['level_2', 'lags_up_to_5'],
exogenous_features = ['z_score'],
index_column= 'int',
no_of_classes= 'two',
primary_models = classification_models,
@@ -167,7 +163,6 @@ def get_lightweight_ensemble_config() -> RawConfig:
own_features = ['level_2' ],
other_features = ['level_2'],
exogenous_features = ['z_score'],
index_column= 'int',
no_of_classes= 'two',
primary_models = classification_models,
config/hashing.py
-1
View File
@@ -17,6 +17,5 @@ def hash_data_config(data_config: dict) -> str:
hash_feature_extractors(data_config['own_features']),
hash_feature_extractors(data_config['other_features']),
hash_feature_extractors(data_config['exogenous_features']),
data_config['index_column'],
data_config['no_of_classes'],
]))
config/sweep_ensemble.yaml
-2
View File
@@ -40,8 +40,6 @@ parameters:
value: True
log_returns:
value: True
index_column:
value: 'int'
primary_models:
distribution: categorical
values:
config/sweep_meta_labeling.yaml
-2
View File
@@ -46,8 +46,6 @@ parameters:
distribution: categorical
log_returns:
value: True
index_column:
value: 'int'
primary_models:
value: ["LogisticRegression_two_class", "LDA", "KNN", "CART", "NB", "AB", "RFC", "StaticMom"]
meta_labeling_models:
config/sweep_primary_models.yaml
-2
View File
@@ -40,8 +40,6 @@ parameters:
value: True
log_returns:
value: True
index_column:
value: 'int'
primary_models:
values: [['LogisticRegression_two_class'], ['SVC'], ['LDA'], ['KNN'], ['CART'], ['MNB'], ['NB'], ['AB'], ['RFC'], ['XGB_two_class'], ['LGBM']]
distribution: categorical
data_loader/load_data.py
+3 -11
View File
@@ -29,7 +29,6 @@ def __load_data(assets: DataCollection,
own_features: list[tuple[str, FeatureExtractor, list[int]]],
other_features: list[tuple[str, FeatureExtractor, list[int]]],
exogenous_features: list[tuple[str, FeatureExtractor, list[int]]],
index_column: Literal['date', 'int'],
no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced'],
) -> tuple[pd.DataFrame, pd.Series, pd.Series]:
"""
@@ -83,15 +82,12 @@ def __load_data(assets: DataCollection,
dfs = pd.concat([df.sort_index().reindex(target_df.index) for df in dfs], axis=1).fillna(0.)
dfs.index = pd.DatetimeIndex(dfs.index)
if index_column == 'int':
dfs.reset_index(drop=True, inplace=True)
df_target_asset_only_returns.reset_index(drop=True, inplace=True)
## Create target
target_col = 'target'
returns_col = target_asset[1] + '_returns'
forward_returns = __create_target_cum_forward_returns(df_target_asset_only_returns, returns_col, forecasting_horizon)
forward_returns.index = pd.DatetimeIndex(dfs.index)
dfs[target_col] = __create_target_classes(dfs, returns_col, forecasting_horizon, no_of_classes)
# we need to drop the last row, because we forward-shift the target (see what happens if you call .shift[-1] on a pd.Series)
dfs = dfs.iloc[:-forecasting_horizon]
@@ -216,7 +212,7 @@ def datasource_to_file(data_source: DataSource) -> str:
def load_only_returns(assets: DataCollection, index_column: Literal['date', 'int'], returns: Literal['price', 'returns']) -> pd.DataFrame:
def load_only_returns(assets: DataCollection, returns: Literal['price', 'returns']) -> pd.DataFrame:
assets_future = [__load_df.remote(
data_source=data_source,
@@ -228,10 +224,6 @@ def load_only_returns(assets: DataCollection, index_column: Literal['date', 'int
dfs = [deduplicate_indexes(df) for df in target_asset_df]
dfs = pd.concat(dfs, axis=1)
# dfs = dfs.applymap(lambda x: np.nan if x == 0 else x)
dfs.index = pd.DatetimeIndex(dfs.index)
if index_column == 'int':
dfs.reset_index(drop=True, inplace=True)
return dfs
run_inference.py
+14 -2
View File
@@ -8,6 +8,7 @@ from config.config import Config, get_dev_config, get_default_ensemble_config, g
from typing import Callable, Optional
from reporting.types import Reporting
from training.training_steps import primary_step, secondary_step
import pandas as pd
import warnings
def run_inference(preload_models:bool, get_config:Callable):
@@ -25,10 +26,21 @@ def __inference(config: Config, primary_models: Optional[Reporting.Training_Step
# 1. Load data, check for validity and process data
X, y, target_returns = load_data(
assets = config.assets,
other_assets = config.other_assets,
exogenous_data = config.exogenous_data,
target_asset = config.target_asset,
load_non_target_asset = config.load_non_target_asset,
log_returns = config.log_returns,
forecasting_horizon = config.forecasting_horizon,
own_features = config.own_features,
other_features = config.other_features,
exogenous_features = config.exogenous_features,
no_of_classes = config.no_of_classes,
)
assert check_data(X, y, config) == True, "Data is not valid. Cancelling Inference."
inference_from = X.index.stop - 2
inference_from: pd.Timestamp = X.index[len(X.index) - 2]
# 2. Train a Primary model with optional metalabeling for each asset
training_step_primary, current_predictions = primary_step(X, y, target_returns, config, reporting, from_index = inference_from, preloaded_training_step = primary_models)
@@ -38,7 +50,7 @@ def __inference(config: Config, primary_models: Optional[Reporting.Training_Step
training_step_secondary = secondary_step(X, y, current_predictions, target_returns, config, reporting, from_index = inference_from, preloaded_training_step = secondary_models)
# 4. Save the models
reporting.asset = Reporting.Asset(ticker=asset, primary=training_step_primary, secondary=training_step_secondary)
reporting.asset = Reporting.Asset(ticker=asset[1], primary=training_step_primary, secondary=training_step_secondary)
return reporting
run_pipeline.py
-1
View File
@@ -59,7 +59,6 @@ def __run_training(config: Config):
own_features = config.own_features,
other_features = config.other_features,
exogenous_features = config.exogenous_features,
index_column = config.index_column,
no_of_classes = config.no_of_classes,
)
assert check_data(X, y, config) == True, "Data is not valid."
run_portfolio_reporting.py
+3 -3
View File
@@ -98,17 +98,17 @@ def create_quantile_weights(predictions: pd.DataFrame, availability: pd.DataFram
row = row.apply(only_select_bottom_top)
no_of_nonzero_predictions = row[row != 0].count()
units = min(1 / no_of_nonzero_predictions, 0.25)
weights.iloc[index] = row * units
weights.loc[index] = row * units
return weights
predictions = pd.read_csv('output/predictions.csv', index_col=0)
predictions.index = pd.DatetimeIndex(predictions.index)
predictions.columns = ['_'.join(col.replace("model_", "").split("_")[:2]) for col in predictions.columns]
first_index = get_first_valid_return_index(predictions[predictions.columns[0]])
predictions = predictions.iloc[first_index:]
predictions.reset_index(drop=True, inplace=True)
close = load_only_returns(data_collections['daily_crypto'], 'date', 'price')
close = load_only_returns(data_collections['daily_crypto'], 'price')
close = close.iloc[first_index:-1]
close.columns = [col.replace("_returns", "") for col in close.columns]
close = close[predictions.columns]
training/meta_labeling.py
+1 -1
View File
@@ -16,7 +16,7 @@ def train_meta_labeling_model(
models: list[tuple[str, Model]],
config: Config,
model_suffix: str,
from_index: Optional[int],
from_index: Optional[pd.Timestamp],
preloaded_models: Optional[list[tuple[str, pd.Series, list[pd.Series]]]] = None
) -> tuple[pd.Series, pd.Series, pd.DataFrame, list[Reporting.Single_Model]]:
training/primary_model.py
+1 -1
View File
@@ -18,7 +18,7 @@ def train_primary_model(
expanding_window: bool,
sliding_window_size: int,
retrain_every: int,
from_index: Optional[int],
from_index: Optional[pd.Timestamp],
scaler: ScalerTypes,
no_of_classes: Literal['two', 'three-balanced', 'three-imbalanced'],
level: str,
training/training_steps.py
+2 -2
View File
@@ -16,7 +16,7 @@ def primary_step(
target_returns: pd.Series,
config: Config,
reporting: Reporting,
from_index: Optional[int],
from_index: Optional[pd.Timestamp],
preloaded_training_step: Optional[Reporting.Training_Step] = None,
) -> tuple[Reporting.Training_Step, pd.DataFrame]:
training_step = Reporting.Training_Step(level='primary')
@@ -78,7 +78,7 @@ def secondary_step(
target_returns:pd.Series,
config: Config,
reporting: Reporting,
from_index: Optional[int],
from_index: Optional[pd.Timestamp],
preloaded_training_step: Optional[Reporting.Training_Step] = None,
) -> Reporting.Training_Step:
training_step = Reporting.Training_Step(level='secondary')
training/walk_forward.py
+17 -17
View File
@@ -15,7 +15,7 @@ def walk_forward_train(
expanding_window: bool,
window_size: int,
retrain_every: int,
from_index: Optional[int],
from_index: Optional[pd.Timestamp],
transformations: list[Transformation],
preloaded_transformations: Optional[list[pd.Series]],
) -> tuple[pd.Series, list[pd.Series]]:
@@ -24,7 +24,7 @@ def walk_forward_train(
transformations_over_time = [pd.Series(index=y.index).rename(t.get_name()) for t in transformations]
first_nonzero_return = max(get_first_valid_return_index(target_returns), get_first_valid_return_index(X.iloc[:,0]), get_first_valid_return_index(y))
train_from = first_nonzero_return + window_size + 1 if from_index is None else from_index
train_from = first_nonzero_return + window_size + 1 if from_index is None else X.index.to_list().index(from_index)
train_till = len(y)
iterations_before_retrain = 0
@@ -35,14 +35,14 @@ def walk_forward_train(
transformations = []
for index in tqdm(range(train_from, train_till)):
train_window_start = first_nonzero_return if expanding_window else index - window_size - 1
train_window_start = X.index[first_nonzero_return] if expanding_window else X.index[index - window_size - 1]
if iterations_before_retrain <= 0 or pd.isna(models_over_time[index-1]):
train_window_end = index - 1
train_window_end = X.index[index - 1]
X_expanding_window = X[first_nonzero_return:train_window_end]
y_expanding_window = y[first_nonzero_return:train_window_end]
X_expanding_window = X[X.index[first_nonzero_return]:train_window_end]
y_expanding_window = y[X.index[first_nonzero_return]:train_window_end]
if preloaded_transformations is not None and len(transformations) > 0:
current_transformations = [transformation_over_time[index] for transformation_over_time in preloaded_transformations]
@@ -66,9 +66,9 @@ def walk_forward_train(
iterations_before_retrain = retrain_every
models_over_time[index] = current_model
models_over_time[X.index[index]] = current_model
for transformation_index, transformation in enumerate(current_transformations):
transformations_over_time[transformation_index][index] = transformation
transformations_over_time[transformation_index][X.index[index]] = transformation
iterations_before_retrain -= 1
@@ -81,12 +81,12 @@ def walk_forward_inference(
X: pd.DataFrame,
expanding_window: bool,
window_size: int,
from_index: Optional[int],
from_index: Optional[pd.Timestamp],
) -> tuple[pd.Series, pd.DataFrame]:
predictions = pd.Series(index=X.index).rename(model_name)
probabilities = pd.DataFrame(index=X.index)
inference_from = max(get_first_valid_return_index(model_over_time), get_first_valid_return_index(X.iloc[:,0])) if from_index is None else from_index
inference_from = max(get_first_valid_return_index(model_over_time), get_first_valid_return_index(X.iloc[:,0])) if from_index is None else X.index.to_list().index(from_index)
inference_till = X.shape[0]
first_model = model_over_time[inference_from]
@@ -98,16 +98,16 @@ def walk_forward_inference(
for index in tqdm(range(inference_from, inference_till)):
train_window_start = inference_from if expanding_window else index - window_size - 1
train_window_start = X.index[inference_from] if expanding_window else X.index[index - window_size - 1]
current_model = model_over_time[index]
current_transformations = [transformation_over_time[index] for transformation_over_time in transformations_over_time]
current_model = model_over_time[X.index[index]]
current_transformations = [transformation_over_time[X.index[index]] for transformation_over_time in transformations_over_time]
if current_model.predict_window_size == 'window_size':
next_timestep = X.iloc[train_window_start:index]
next_timestep = X.loc[train_window_start:X.index[index]]
else:
# we need to get a Dataframe out of it, since the transformation step always expects a 2D array, but it's equivalent to X.iloc[index]
next_timestep = X.iloc[index:index+1]
next_timestep = X.loc[X.index[index]:X.index[index]]
for transformation in current_transformations:
next_timestep = transformation.transform(next_timestep)
@@ -115,9 +115,9 @@ def walk_forward_inference(
next_timestep = next_timestep.to_numpy()
prediction, probs = current_model.predict(next_timestep)
predictions[index] = prediction
predictions[X.index[index]] = prediction
if len(probabilities.columns) != len(probs):
probabilities = probabilities.reindex(columns = ["prob_" + str(num) for num in range(0, len(probs.T))])
probabilities.iloc[index] = probs
probabilities.loc[X.index[index]] = probs
return predictions, probabilities
utils/evaluate.py
+1 -2
View File
@@ -36,8 +36,7 @@ def evaluate_predictions(
discretize: bool = False,
) -> pd.Series:
# ignore the predictions until we see a non-zero returns (and definitely skip the first sliding_window_size)
first_nonzero_return = max(get_first_valid_return_index(target_returns), get_first_valid_return_index(y_pred))
evaluate_from = first_nonzero_return + 1
evaluate_from = max(get_first_valid_return_index(target_returns), get_first_valid_return_index(y_pred))
target_returns = pd.Series(target_returns[evaluate_from:])
y_pred = pd.Series(y_pred[evaluate_from:])