4.7 MiB
4.7 MiB
In [ ]:
import sys
import os
import warnings
from pathlib import Path
# ---------------------------------------------------------------------------
# 1) SET PROJECT ROOT AND UPDATE PATH/WORKING DIRECTORY
# ---------------------------------------------------------------------------
project_root = Path.cwd().parent.parent # Adjust if your notebook is in notebooks/time_series
sys.path.append(str(project_root))
os.chdir(str(project_root))
warnings.filterwarnings("ignore")
import warnings
warnings.filterwarnings("ignore")
import pandas as pd
import numpy as np
import MetaTrader5 as mt5
# If using vectorbt
import vectorbt as vbt
# Our modules
from data.data_loader_mt5 import get_data_mt5
from features.feature_engineering import add_all_ta_features
from models.model_training import (
select_features_rf_reg,
walk_forward_splits
)
from backtests.simple_backtest import simulate_trading, calculate_sharpe_ratio
# Sklearn / Models
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import accuracy_score
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.svm import SVC
from xgboost import XGBClassifier
from lightgbm import LGBMClassifier
# Suppose you have a multi-bar labeling function
from features.labeling_schemes import create_labels_multi_bar # or define inline
from sklearn.naive_bayes import GaussianNB, BernoulliNB
###########################################################
# 1) Global Variables
###########################################################
symbol = "US500"
timeframe = mt5.TIMEFRAME_H4
n_bars = 10000
start_pos = 0
horizon=5 # Horizon for multi-bar labeling
threshold=0.005 # Threshold for multi-bar labeling
###########################################################
# 2) Data Loading & Basic Feature Engineering
###########################################################
if not mt5.initialize():
print("Failed to initialize MT5")
else:
# Now using your global variables
data = get_data_mt5(symbol=symbol, timeframe=timeframe, n_bars=n_bars, start_pos=start_pos)
mt5.shutdown()
df = add_all_ta_features(data)
###########################################################
# 2) Multi-Bar Labeling Function
###########################################################
def create_labels_multi_bar(df, horizon=5, threshold=0.005):
"""
Creates classification labels for multi-bar horizon:
+1 if future return >= threshold
-1 if future return <= -threshold
0 otherwise
df must have a 'close' column.
Returns a new DataFrame with:
'future_return_h' and 'multi_bar_label'
"""
df_copy = df.copy()
# 1) Horizon-based future returns
df_copy["future_return_h"] = df_copy["close"].pct_change(periods=horizon).shift(-horizon)
# 2) Classification labels
df_copy["multi_bar_label"] = 0
df_copy.loc[df_copy["future_return_h"] >= threshold, "multi_bar_label"] = 1
df_copy.loc[df_copy["future_return_h"] <= -threshold, "multi_bar_label"] = -1
# 3) Drop rows where future_return_h is NaN
df_copy.dropna(subset=["future_return_h"], inplace=True)
return df_copy
df_lbl = create_labels_multi_bar(df, horizon=horizon, threshold=threshold)
# Prepare X, y
X = df_lbl.drop(columns=["multi_bar_label", "future_return_h"])
y = df_lbl["multi_bar_label"]
###########################################################
# 3) Walk-Forward Splits
###########################################################
folds = walk_forward_splits(X, y, n_splits=3)
print(f"Number of folds created: {len(folds)}")
###########################################################
# 4) Define Classification Models
###########################################################
models = {
"RandomForestClassifier": RandomForestClassifier(n_estimators=100, random_state=42),
"GradientBoostingClassifier": GradientBoostingClassifier(n_estimators=100, learning_rate=0.1, max_depth=5, random_state=42),
"SVC": SVC(C=1.0, kernel='rbf', probability=True),
"XGBClassifier": XGBClassifier(n_estimators=100, learning_rate=0.1, random_state=42),
"LGBMClassifier": LGBMClassifier(n_estimators=100, learning_rate=0.1, random_state=42),
"GaussianNB": GaussianNB(), # <-- Added Bayesian Classification Model
"BernoulliNB": BernoulliNB() # <-- Another Bayesian Model (good for binary data)
}
###########################################################
# 5) Loop Over Folds + Simple Backtest
###########################################################
fold_results = {}
for fold_i, (X_train_fold, y_train_fold, X_test_fold, y_test_fold) in enumerate(folds, start=1):
print(f"\n===== Fold {fold_i} =====")
# We must shift labels from [-1, 0, 1] to [0, 1, 2] for XGBoost & co.
# SHIFT: -1 -> 0, 0 -> 1, +1 -> 2
y_train_fold_shifted = y_train_fold + 1
y_test_fold_shifted = y_test_fold + 1
# Feature selection with a classifier
rf_for_fs = RandomForestClassifier(n_estimators=100, random_state=42)
# Use the SHIFTED y_train for feature selection
X_train_sel, selected_idx = select_features_rf_reg(
X_train_fold, y_train_fold_shifted, estimator=rf_for_fs, max_features=20
)
feats = X_train_fold.columns[selected_idx]
print(f"Selected features for Fold {fold_i}: {feats.tolist()}")
X_test_sel = X_test_fold[feats]
# Scale
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train_sel)
X_test_scaled = scaler.transform(X_test_sel)
fold_results[fold_i] = {}
for model_name, model in models.items():
# 1) Fit on SHIFTED y
model.fit(X_train_scaled, y_train_fold_shifted)
# 2) Predict SHIFTED labels
preds_shifted = model.predict(X_test_scaled)
# 3) Shift back: 0->-1, 1->0, 2->+1
preds = preds_shifted - 1
# Evaluate Accuracy on the unshifted test labels
acc = accuracy_score(y_test_fold, preds)
# Convert classification => signals
signals = preds # signals in {-1, 0, +1}
# Align with the test portion
df_test_fold = df_lbl.loc[X_test_fold.index].copy()
# Simple backtest with cost
daily_returns, total_return = simulate_trading(signals, df_test_fold, cost=0.0002)
sr = calculate_sharpe_ratio(np.array(daily_returns))
fold_results[fold_i][model_name] = {
"Accuracy": acc,
"TotalReturn": total_return,
"Sharpe": sr
}
###########################################################
# 6) Print Results
###########################################################
for fold_i, model_dict in fold_results.items():
print(f"\n=== Fold {fold_i} Results ===")
for model_name, stats in model_dict.items():
acc = stats["Accuracy"]
ret = stats["TotalReturn"]
sr = stats["Sharpe"]
print(f"{model_name}: ACC={acc:.2f}, Return={ret:.2f}%, Sharpe={sr:.2f}")
Number of folds created: 3 ===== Fold 1 ===== Selected features for Fold 1: ['volume_vpt', 'volatility_atr', 'volatility_kcw', 'volatility_ui', 'volatility_dcw', 'momentum_tsi', 'momentum_pvo_signal', 'volume_nvi', 'volatility_bbw', 'volume_adi', 'trend_adx_pos', 'momentum_ppo_signal', 'trend_macd_signal', 'volume_fi', 'trend_adx', 'momentum_rsi', 'trend_adx_neg', 'trend_kst', 'volume_cmf', 'tick_volume']
File "c:\Users\moham\miniconda3\envs\ml\Lib\site-packages\joblib\externals\loky\backend\context.py", line 257, in _count_physical_cores
cpu_info = subprocess.run(
^^^^^^^^^^^^^^^
File "c:\Users\moham\miniconda3\envs\ml\Lib\subprocess.py", line 548, in run
with Popen(*popenargs, **kwargs) as process:
^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "c:\Users\moham\miniconda3\envs\ml\Lib\subprocess.py", line 1026, in __init__
self._execute_child(args, executable, preexec_fn, close_fds,
File "c:\Users\moham\miniconda3\envs\ml\Lib\subprocess.py", line 1538, in _execute_child
hp, ht, pid, tid = _winapi.CreateProcess(executable, args,
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
[LightGBM] [Info] Auto-choosing col-wise multi-threading, the overhead of testing was 0.013866 seconds. You can set `force_col_wise=true` to remove the overhead. [LightGBM] [Info] Total Bins 5100 [LightGBM] [Info] Number of data points in the train set: 2498, number of used features: 20 [LightGBM] [Info] Start training from score -1.509698 [LightGBM] [Info] Start training from score -0.706852 [LightGBM] [Info] Start training from score -1.252363 ===== Fold 2 ===== Selected features for Fold 2: ['volatility_kcw', 'volatility_atr', 'volatility_ui', 'volatility_dcw', 'volume_vpt', 'volatility_bbw', 'momentum_pvo_signal', 'volume_cmf', 'trend_mass_index', 'trend_adx', 'momentum_tsi', 'momentum_ppo_signal', 'trend_kst', 'trend_stc', 'trend_adx_pos', 'trend_macd', 'trend_macd_signal', 'trend_kst_diff', 'trend_kst_sig', 'momentum_uo'] [LightGBM] [Info] Auto-choosing col-wise multi-threading, the overhead of testing was 0.012360 seconds. You can set `force_col_wise=true` to remove the overhead. [LightGBM] [Info] Total Bins 5100 [LightGBM] [Info] Number of data points in the train set: 4996, number of used features: 20 [LightGBM] [Info] Start training from score -1.572306 [LightGBM] [Info] Start training from score -0.679233 [LightGBM] [Info] Start training from score -1.253764 ===== Fold 3 ===== Selected features for Fold 3: ['volatility_kcw', 'volatility_atr', 'volatility_ui', 'volatility_dcw', 'volume_vpt', 'momentum_pvo_signal', 'volatility_bbw', 'trend_adx', 'trend_mass_index', 'trend_adx_pos', 'volume_cmf', 'momentum_tsi', 'trend_stc', 'trend_kst', 'trend_kst_sig', 'volume_adi', 'momentum_ppo_signal', 'volume_fi', 'trend_macd_signal', 'trend_adx_neg']
In [1]:
# Code 2: Hyperparameter Tuning for Chosen Classification Model
import sys
import os
import warnings
from pathlib import Path
project_root = Path.cwd().parent.parent # Adjust if needed
sys.path.append(str(project_root))
os.chdir(str(project_root))
warnings.filterwarnings("ignore")
import pandas as pd
import numpy as np
import MetaTrader5 as mt5
import joblib
# Sklearn / Models
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import TimeSeriesSplit, RandomizedSearchCV
from sklearn.metrics import accuracy_score, make_scorer
from sklearn.pipeline import Pipeline
# Your modules
from data.data_loader_mt5 import get_data_mt5
from features.feature_engineering import add_all_ta_features
# Suppose you have a multi-bar labeling function
from features.labeling_schemes import create_labels_multi_bar # or define inline
###########################################################
# 1) DATA LOADING & FEATURE ENGINEERING
###########################################################
if not mt5.initialize():
print("Failed to initialize MT5")
else:
# Fetch 2000 bars from an earlier period for training
data = get_data_mt5(symbol="US30", timeframe=mt5.TIMEFRAME_H4, n_bars=5000, start_pos=5000)
mt5.shutdown()
df = add_all_ta_features(data)
# Create classification labels
# e.g., horizon=5, threshold=0.005 => ±0.5% over 5 bars
df_lbl = create_labels_multi_bar(df, horizon=5, threshold=0.005)
# Now we have columns: 'future_return_h' and 'multi_bar_label' in df_lbl
X_full = df_lbl.drop(columns=["multi_bar_label", "future_return_h"])
y_full = df_lbl["multi_bar_label"]
# SHIFT LABELS from [-1,0,+1] => [0,1,2]
# so the classifier won't complain about negative labels
y_full_shifted = y_full + 1 # -1->0, 0->1, +1->2
print("Unique classes in y_full:", y_full.unique())
print("Unique classes in y_full_shifted:", y_full_shifted.unique())
# Ensure chronological order if needed
# X_full = X_full.sort_index()
# y_full_shifted = y_full_shifted.loc[X_full.index]
###########################################################
# 2) DEFINE YOUR TRAIN PORTION
###########################################################
# e.g., first 80% for tuning
split_idx = int(len(X_full)*0.8)
X_tune = X_full.iloc[:split_idx].copy()
y_tune_shifted = y_full_shifted.iloc[:split_idx].copy()
print(f"Tuning portion size: {len(X_tune)}")
###########################################################
# 3) TIME-BASED CV (TimeSeriesSplit)
###########################################################
tscv = TimeSeriesSplit(n_splits=3)
# We'll define a scoring for classification
# e.g. "accuracy"
scorer = make_scorer(accuracy_score)
###########################################################
# 4) BUILD A PIPELINE
###########################################################
pipeline = Pipeline([
("scaler", StandardScaler()),
("clf", RandomForestClassifier(random_state=42))
])
###########################################################
# 5) DEFINE PARAM DISTRIBUTIONS FOR RandomForestClassifier
###########################################################
param_distributions = {
"clf__n_estimators": [100, 200, 300],
"clf__max_depth": [None, 5, 10, 15],
"clf__min_samples_split": [2, 5, 10],
"clf__max_features": ["auto", "sqrt", 0.5]
}
###########################################################
# 6) SET UP RandomizedSearchCV
###########################################################
random_search = RandomizedSearchCV(
estimator=pipeline,
param_distributions=param_distributions,
n_iter=10, # how many random combos
scoring=scorer, # 'accuracy' metric
cv=tscv, # time-based folds
random_state=42,
n_jobs=-1,
verbose=2
)
###########################################################
# 7) FIT ON TUNING PORTION
###########################################################
random_search.fit(X_tune, y_tune_shifted)
print("Best params:", random_search.best_params_)
print("Best score (accuracy):", random_search.best_score_)
best_estimator = random_search.best_estimator_
###########################################################
# 8) SAVE THE BEST ESTIMATOR
###########################################################
joblib.dump(best_estimator, "models/saved_models/best_rf_mb_pipeline.pkl")
print("Saved best estimator to 'best_rf_mb_pipeline.pkl'")
Unique classes in y_full: [-1 0 1]
Unique classes in y_full_shifted: [0 1 2]
Tuning portion size: 3996
Fitting 3 folds for each of 10 candidates, totalling 30 fits
Best params: {'clf__n_estimators': 100, 'clf__min_samples_split': 2, 'clf__max_features': 0.5, 'clf__max_depth': 5}
Best score (accuracy): 0.3720387053720387
Saved best estimator to 'best_rf_mb_pipeline.pkl'
In [ ]:
# ----------------------------------------------------------------------------
# 0) SETUP
# ----------------------------------------------------------------------------
import sys
import os
import warnings
from pathlib import Path
project_root = Path.cwd().parent.parent # Adjust if needed
sys.path.append(str(project_root))
os.chdir(str(project_root))
warnings.filterwarnings("ignore")
import pandas as pd
import numpy as np
import MetaTrader5 as mt5
import joblib
import optuna
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import TimeSeriesSplit
from sklearn.pipeline import Pipeline
from sklearn.metrics import accuracy_score
from data.data_loader_mt5 import get_data_mt5
from features.feature_engineering import add_all_ta_features
from features.labeling_schemes import create_labels_multi_bar
import optuna.visualization as ov
# ----------------------------------------------------------------------------
# 1) LOAD DATA
# ----------------------------------------------------------------------------
symbol = "DE40"
timeframe = mt5.TIMEFRAME_M30
n_bars = 10000
start_pos = 5000
if not mt5.initialize():
raise RuntimeError("Failed to initialize MT5")
data = get_data_mt5(symbol=symbol, timeframe=timeframe, n_bars=n_bars, start_pos=start_pos)
mt5.shutdown()
print("✅ Data shape:", data.shape)
# ----------------------------------------------------------------------------
# 2) FEATURE ENGINEERING
# ----------------------------------------------------------------------------
horizon = 5
threshold = 0.005 # ✅ Based on your good threshold scan (balance up/neutral/down)
df = add_all_ta_features(data)
df_lbl = create_labels_multi_bar(df, horizon=horizon, threshold=threshold)
X_full = df_lbl.drop(columns=["multi_bar_label", "future_return_h"])
y_full = df_lbl["multi_bar_label"]
y_full_shifted = y_full + 1 # Shift (-1,0,+1) ➔ (0,1,2)
# ----------------------------------------------------------------------------
# 3) TRAINING SETUP
# ----------------------------------------------------------------------------
split_idx = int(len(X_full) * 0.8)
X_tune, y_tune = X_full.iloc[:split_idx], y_full_shifted.iloc[:split_idx]
print(f"Tuning samples: {X_tune.shape[0]}")
# ----------------------------------------------------------------------------
# 4) FEATURE SELECTION
# ----------------------------------------------------------------------------
fs_model = RandomForestClassifier(n_estimators=100, random_state=42)
fs_model.fit(X_tune, y_tune)
important_features = pd.Series(fs_model.feature_importances_, index=X_tune.columns)
top_features = important_features.nlargest(20).index.tolist()
X_tune = X_tune[top_features]
print(f"✅ Selected Top Features: {top_features}")
# ----------------------------------------------------------------------------
# 5) DEFINE OPTUNA OBJECTIVE (with TimeSeriesSplit)
# ----------------------------------------------------------------------------
tscv = TimeSeriesSplit(n_splits=3)
def objective(trial):
params = {
"n_estimators": trial.suggest_int("n_estimators", 100, 1000, step=100),
"max_depth": trial.suggest_int("max_depth", 5, 30),
"min_samples_split": trial.suggest_int("min_samples_split", 2, 20),
"max_features": trial.suggest_categorical("max_features", ["sqrt", "log2", 0.5, 0.8, None]),
"bootstrap": trial.suggest_categorical("bootstrap", [True, False])
}
model = Pipeline([
("scaler", StandardScaler()),
("rf", RandomForestClassifier(**params, random_state=42))
])
scores = []
for train_idx, test_idx in tscv.split(X_tune):
X_train_fold, X_test_fold = X_tune.iloc[train_idx], X_tune.iloc[test_idx]
y_train_fold, y_test_fold = y_tune.iloc[train_idx], y_tune.iloc[test_idx]
model.fit(X_train_fold, y_train_fold)
preds = model.predict(X_test_fold)
scores.append(accuracy_score(y_test_fold, preds))
return np.mean(scores)
# ----------------------------------------------------------------------------
# 6) RUN OPTUNA
# ----------------------------------------------------------------------------
study = optuna.create_study(direction="maximize")
study.optimize(objective, n_trials=50, timeout=1800)
print("\n✅ Best Trial:", study.best_trial)
# ----------------------------------------------------------------------------
# 7) RETRAIN BEST MODEL ON FULL TUNING DATA
# ----------------------------------------------------------------------------
best_params = study.best_params
final_model = Pipeline([
("scaler", StandardScaler()),
("rf", RandomForestClassifier(**best_params, random_state=42))
])
final_model.fit(X_tune, y_tune)
# ----------------------------------------------------------------------------
# 8) SAVE MODEL
# ----------------------------------------------------------------------------
save_dir = Path("models/saved_models")
save_dir.mkdir(parents=True, exist_ok=True)
model_filename = f"best_rf_multibar_{symbol}_h{horizon}_thr{threshold:.4f}.pkl"
model_path = save_dir / model_filename
joblib.dump(final_model, model_path)
print(f"\n✅ Model saved to {model_path}")
# ----------------------------------------------------------------------------
# 9) OPTUNA VISUALIZATIONS
# ----------------------------------------------------------------------------
ov.plot_optimization_history(study).show()
ov.plot_param_importances(study).show()
ov.plot_parallel_coordinate(study).show()
✅ Data shape: (10000, 7) Tuning samples: 7996
[I 2025-04-20 16:28:49,275] A new study created in memory with name: no-name-93ed0bff-0433-4144-873f-c81f7749fd18
✅ Selected Top Features: ['volume_vpt', 'volatility_atr', 'volatility_kcw', 'volume_obv', 'volatility_dcw', 'momentum_pvo_signal', 'trend_adx', 'momentum_pvo_hist', 'volatility_bbw', 'volume_adi', 'trend_kst', 'momentum_pvo', 'trend_mass_index', 'volume_sma_em', 'trend_kst_diff', 'tick_volume', 'volume_nvi', 'trend_stc', 'volatility_ui', 'trend_macd_signal']
In [ ]:
import sys
import os
import warnings
from pathlib import Path
# ---------------------------------------------------------------------------
# 1) SET PROJECT ROOT AND UPDATE PATH/WORKING DIRECTORY
# ---------------------------------------------------------------------------
project_root = Path.cwd().parent.parent
sys.path.append(str(project_root))
os.chdir(str(project_root))
warnings.filterwarnings("ignore")
import pandas as pd
import numpy as np
import MetaTrader5 as mt5
import vectorbt as vbt
import joblib
# Our modules
from data.data_loader import get_data_mt5
from features.feature_engineering import add_all_ta_features
from features.labeling_schemes import create_labels_multi_bar # multi-bar labeling
# Sklearn
from sklearn.metrics import accuracy_score
###########################################################
# 1) DATA LOADING & FEATURE ENGINEERING
###########################################################
if not mt5.initialize():
print("Failed to initialize MT5")
else:
# Fetch 5000 most recent bars for backtesting
data = get_data_mt5(symbol="US30", timeframe=mt5.TIMEFRAME_H4, n_bars=5000, start_pos=0)
mt5.shutdown()
# Add technical features
df = add_all_ta_features(data)
# Create multi-bar classification labels (e.g., horizon=5, threshold=0.005)
df_lbl = create_labels_multi_bar(df, horizon=5, threshold=0.005)
# Separate features and labels
X = df_lbl.drop(columns=["multi_bar_label", "future_return_h"])
y = df_lbl["multi_bar_label"] # in {-1, 0, +1}
y_shifted = y + 1 # in {0, 1, 2} for classifier (kept for reference)
###########################################################
# 2) LOAD PRE-TRAINED CLASSIFICATION MODEL (NO RETRAINING)
###########################################################
best_pipeline = joblib.load("models/saved_models/best_rf_mb_pipeline.pkl")
print("Loaded best classification model from 'best_rf_mb_pipeline.pkl'")
# 1) Identify columns used during training
trained_columns = best_pipeline["scaler"].feature_names_in_ # adjust if your pipeline step name differs
# 2) Subset X to match these columns
X_test = X[trained_columns]
# 3) Predict on new data (model expects labels in {0,1,2})
preds_shifted = best_pipeline.predict(X_test)
# Convert predictions back to {-1, 0, +1}
preds = preds_shifted - 1
# Accuracy (align truth to prediction rows)
accuracy = accuracy_score(y.loc[X_test.index], preds)
print(f"\nOut-of-Sample Accuracy: {accuracy:.4f}")
###########################################################
# 3) BACKTEST via target exposure (-1, 0, +1) << UPDATED
###########################################################
# predictions already in {-1, 0, +1}
exposure = pd.Series(preds.astype(float), index=X_test.index)
# Align prices exactly to prediction rows (no padding needed)
close = df_lbl.loc[X_test.index, "close"]
# Optional: trade on next bar to avoid look-ahead
execution_lag = 1 # set to 0 if you prefer same-bar execution
if execution_lag > 0:
exposure = exposure.shift(execution_lag).fillna(0.0)
fees = 0.0002 # 0.02% transaction cost per trade
pf = vbt.Portfolio.from_orders(
close=close,
size=exposure, # -1 short, 0 flat, +1 long
size_type='targetpercent',
init_cash=10000,
freq='4H',
fees=fees
)
total_return = pf.total_return()
sharpe_ratio = pf.sharpe_ratio()
print("\nFull Backtest Results:")
print(f"Accuracy={accuracy:.4f}, Return={total_return:.2f}%, Sharpe={sharpe_ratio:.2f}")
print(pf.stats())
# Optional: Plot the backtest
fig = pf.plot()
fig.show()
✅ Loaded model from models\saved_models\best_rf_multibar_US30_h5_thr0.0050.pkl ✅ Data shape: (5000, 7) ✅ Out-of-Sample Accuracy: 0.5363 ✅ Full Backtest Results: Accuracy=0.54, Return=0.13%, Sharpe=0.39 Start 2022-01-20 00:00:00 End 2025-04-17 00:00:00 Period 832 days 12:00:00 Start Value 10000.0 End Value 11276.44095 Total Return [%] 12.76441 Benchmark Return [%] 13.566871 Max Gross Exposure [%] 100.0 Total Fees Paid 83.407265 Max Drawdown [%] 17.035042 Max Drawdown Duration 316 days 04:00:00 Total Trades 20 Total Closed Trades 20 Total Open Trades 0 Open Trade PnL 0.0 Win Rate [%] 55.0 Best Trade [%] 8.638011 Worst Trade [%] -6.172178 Avg Winning Trade [%] 3.533494 Avg Losing Trade [%] -2.812692 Avg Winning Trade Duration 24 days 01:49:05.454545454 Avg Losing Trade Duration 27 days 08:26:40 Profit Factor 1.478091 Expectancy 63.822048 Sharpe Ratio 0.388668 Calmar Ratio 0.317473 Omega Ratio 1.035237 Sortino Ratio 0.567976 dtype: object
[Data output - unsupported data type map[string]interface {} for mime type application/vnd.plotly.v1+json]
In [1]:
# ---------------------------------------------------------------------------
# 1) SET PROJECT ROOT AND UPDATE PATH/WORKING DIRECTORY
import sys
import os
import warnings
from pathlib import Path
# Set paths
project_root = Path.cwd().parent.parent
sys.path.append(str(project_root))
os.chdir(str(project_root))
warnings.filterwarnings("ignore")
import pandas as pd
import numpy as np
import MetaTrader5 as mt5
import vectorbt as vbt
import joblib
# Our modules
from data.data_loader_mt5 import get_data_mt5
from features.feature_engineering import add_all_ta_features
from features.labeling_schemes import create_labels_multi_bar # multi-bar labeling
# Sklearn
from sklearn.metrics import accuracy_score
###########################################################
# 1) DATA LOADING & FEATURE ENGINEERING
###########################################################
if not mt5.initialize():
print("Failed to initialize MT5")
else:
# Fetch 5000 most recent bars for backtesting
data = get_data_mt5(symbol="US30", timeframe=mt5.TIMEFRAME_H4, n_bars=5000, start_pos=0)
mt5.shutdown()
# Add technical features
df = add_all_ta_features(data)
# Create multi-bar classification labels (e.g., horizon=5, threshold=0.005)
df_lbl = create_labels_multi_bar(df, horizon=5, threshold=0.005)
# Separate features and labels
X = df_lbl.drop(columns=["multi_bar_label", "future_return_h"])
y = df_lbl["multi_bar_label"]
# Shift labels from [-1,0,+1] → [0,1,2] for classifier
y_shifted = y + 1
###########################################################
# 2) LOAD PRE-TRAINED CLASSIFICATION MODEL (NO RETRAINING)
###########################################################
best_pipeline = joblib.load("models/saved_models/best_rf_mb_pipeline.pkl")
print("Loaded best classification model from 'best_rf_mb_pipeline.pkl'")
# 1) Identify columns used during training
trained_columns = best_pipeline["scaler"].feature_names_in_ # adjust if your pipeline step name differs
# 2) Subset X to match these columns
X_test = X[trained_columns]
# 3) Predict on new data
preds_shifted = best_pipeline.predict(X_test)
# Convert predictions back to [-1, 0, +1]
preds = preds_shifted - 1
# Compute accuracy against true labels (also shifted back)
y_true_unshifted = y_shifted - 1
accuracy = accuracy_score(y_true_unshifted, preds)
print(f"\nOut-of-Sample Accuracy: {accuracy:.4f}")
###########################################################
# 3) CONVERT PREDICTIONS TO SIGNALS & BACKTEST
###########################################################
# +1 => buy, -1 => sell, 0 => no position
signals = preds
print("\nRunning Full Backtest on the Last 2000 Bars...")
close_prices = df_lbl["close"]
if len(signals) < len(close_prices):
signals = np.append(signals, [0] * (len(close_prices) - len(signals)))
signals_s = pd.Series(signals, index=close_prices.index)
fees = 0.0002 # 0.02% transaction cost per trade
pf = vbt.Portfolio.from_signals(
close_prices,
entries=signals_s > 0,
exits=signals_s < 0,
init_cash=10000,
freq='4H',
fees=fees
)
total_return = pf.total_return()
sharpe_ratio = pf.sharpe_ratio()
print("\nFull Backtest Results:")
print(f"Accuracy={accuracy:.2f}, Return={total_return:.2f}%, Sharpe={sharpe_ratio:.2f}")
print(pf.stats())
# Optional: Plot the backtest
fig = pf.plot()
fig.show()
Loaded best classification model from 'best_rf_mb_pipeline.pkl' Out-of-Sample Accuracy: 0.4981 Running Full Backtest on the Last 2000 Bars... Full Backtest Results: Accuracy=0.50, Return=0.05%, Sharpe=0.33 Start 2022-01-20 00:00:00 End 2025-04-17 00:00:00 Period 832 days 12:00:00 Start Value 10000.0 End Value 10457.629442 Total Return [%] 4.576294 Benchmark Return [%] 13.566871 Max Gross Exposure [%] 100.0 Total Fees Paid 241.883964 Max Drawdown [%] 9.42178 Max Drawdown Duration 661 days 00:00:00 Total Trades 59 Total Closed Trades 59 Total Open Trades 0 Open Trade PnL 0.0 Win Rate [%] 62.711864 Best Trade [%] 4.553492 Worst Trade [%] -2.74492 Avg Winning Trade [%] 0.702686 Avg Losing Trade [%] -0.960206 Avg Winning Trade Duration 0 days 22:29:11.351351351 Avg Losing Trade Duration 1 days 02:32:43.636363636 Profit Factor 1.210019 Expectancy 7.756431 Sharpe Ratio 0.328969 Calmar Ratio 0.210283 Omega Ratio 1.083233 Sortino Ratio 0.465464 dtype: object
[Data output - unsupported data type map[string]interface {} for mime type application/vnd.plotly.v1+json]
In [2]:
import plotly.graph_objects as go
# ----------------------------------------------------------------------------
# Prepare Actual Close Series
# ----------------------------------------------------------------------------
actual_close = close_prices
# ----------------------------------------------------------------------------
# Build Predicted Close Series
# ----------------------------------------------------------------------------
# Our model predicts UP (1), NEUTRAL (0), DOWN (-1)
# Let's map these to expected price movement over the next `horizon` bars
pred_movement = preds # Already -1, 0, +1
# Assume simple model: if UP predicted => price will rise (linearly), DOWN => fall
# Simulate predicted future price based on last close
predicted_close = actual_close.copy()
for idx in range(len(predicted_close) - horizon):
window = pred_movement[idx: idx + horizon]
predicted_close.iloc[idx + horizon] = predicted_close.iloc[idx] * (1 + 0.001 * window.sum())
# Cut predicted series to align properly
predicted_close = predicted_close.shift(-horizon)
# ----------------------------------------------------------------------------
# Plot
# ----------------------------------------------------------------------------
fig = go.Figure()
# Actual price
fig.add_trace(go.Scatter(
x=actual_close.index,
y=actual_close,
mode='lines',
name='Actual Close',
line=dict(color='black')
))
# Predicted price
fig.add_trace(go.Scatter(
x=predicted_close.index,
y=predicted_close,
mode='lines',
name='Predicted Close',
line=dict(color='orange')
))
# Layout
fig.update_layout(
title=f"📈 Actual vs Predicted Close Price (horizon={horizon})",
xaxis_title="Date",
yaxis_title="Price",
template="plotly_white",
legend=dict(x=0, y=1, bgcolor="rgba(255,255,255,0)", bordercolor="Black"),
height=700
)
fig.show()
[Data output - unsupported data type map[string]interface {} for mime type application/vnd.plotly.v1+json]