File size: 7,883 Bytes
fc115d5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 | #!/usr/bin/env python3
"""
Advanced Training Pipeline (Phase 11.5)
Implements Walk-Forward Cross Validation to prevent overfitting and
simulate realistic deployment conditions.
Pipeline:
1. Load historical data (2+ years)
2. Split into rolling windows (e.g., 6 months train, 2 months test)
3. Train PPO sequentially on each train window
4. Evaluate on each test window
5. Save models for each window and a final combined model
"""
import sys
import os
import argparse
import logging
import numpy as np
import pandas as pd
from pathlib import Path
from datetime import datetime
from dateutil.relativedelta import relativedelta
from stable_baselines3 import PPO
from stable_baselines3.common.vec_env import DummyVecEnv, VecNormalize
from stable_baselines3.common.monitor import Monitor
from stable_baselines3.common.callbacks import EvalCallback
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../..')))
from src.env.mtf_env import MultiTimeframeTradingEnv, create_mtf_env
from src.backtest.data_loader import DataLoader
logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(name)s] %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class WalkForwardTrainer:
def __init__(
self,
symbol: str = "BTCUSDT",
train_months: int = 6,
test_months: int = 2,
total_days: int = 730, # 2 years
):
self.symbol = symbol
self.train_months = train_months
self.test_months = test_months
self.total_days = total_days
self.model_dir = Path("./data/models/walkforward")
self.model_dir.mkdir(parents=True, exist_ok=True)
# Hyperparams for rapid adaptation
self.ppo_kwargs = {
'policy': 'MlpPolicy',
'learning_rate': 3e-4,
'n_steps': 2048,
'batch_size': 128,
'n_epochs': 10,
'gamma': 0.99,
'gae_lambda': 0.95,
'clip_range': 0.2,
'ent_coef': 0.01,
'vf_coef': 0.5,
'max_grad_norm': 0.5,
'policy_kwargs': {
'net_arch': {'pi': [256, 256], 'vf': [256, 256]}
}
}
def fetch_data(self) -> pd.DataFrame:
from src.data.multi_asset_fetcher import MultiAssetDataFetcher
fetcher = MultiAssetDataFetcher()
end_date = datetime.now()
start_date = end_date - pd.Timedelta(days=self.total_days)
logger.info(f"Fetching {self.symbol} data from {start_date.date()} to {end_date.date()}...")
df = fetcher.fetch_asset(
symbol=self.symbol,
interval="1h",
days=self.total_days,
)
return df
def create_windows(self, df: pd.DataFrame):
windows = []
if df.empty:
return windows
start_ts = df['timestamp'].iloc[0]
end_ts = df['timestamp'].iloc[-1]
current_start = start_ts
while True:
# Calculate window boundaries
train_end = current_start + pd.Timedelta(days=30 * self.train_months)
test_end = train_end + pd.Timedelta(days=30 * self.test_months)
if test_end > end_ts:
# We've reached the end of the data
break
train_df = df[(df['timestamp'] >= current_start) & (df['timestamp'] < train_end)].copy()
test_df = df[(df['timestamp'] >= train_end) & (df['timestamp'] < test_end)].copy()
# Need a minimum amount of data to compute features
if len(train_df) > 500 and len(test_df) > 100:
windows.append((train_df, test_df))
# Roll forward by 'test_months'
current_start = current_start + pd.Timedelta(days=30 * self.test_months)
logger.info(f"Created {len(windows)} walk-forward windows.")
return windows
def run(self):
df = self.fetch_data()
windows = self.create_windows(df)
if not windows:
logger.error("No windows created. Not enough data.")
return
model = None
vec_env = None
all_test_metrics = []
for i, (train_df, test_df) in enumerate(windows):
logger.info("=" * 60)
logger.info(f"Walk-Forward Window {i+1}/{len(windows)}")
logger.info(f"Train: {train_df['timestamp'].iloc[0].date()} to {train_df['timestamp'].iloc[-1].date()} ({len(train_df)} bars)")
logger.info(f"Test : {test_df['timestamp'].iloc[0].date()} to {test_df['timestamp'].iloc[-1].date()} ({len(test_df)} bars)")
logger.info("=" * 60)
# Create sub-environments
train_envs = create_mtf_env(train_df, n_envs=4, randomize_start=True)
eval_envs = create_mtf_env(test_df, n_envs=1, randomize_start=False)
if model is None:
logger.info("Initializing base model...")
model = PPO(env=train_envs, verbose=0, **self.ppo_kwargs)
else:
logger.info("Continuing training from previous window...")
model.set_env(train_envs)
eval_callback = EvalCallback(
eval_envs,
best_model_save_path=f"./data/models/walkforward/window_{i}",
log_path=f"./data/models/walkforward/logs_{i}",
eval_freq=10000,
deterministic=True,
render=False
)
# Curriculum step size - train for ~100k steps per window
steps_per_window = min(150000, len(train_df) * 10)
logger.info(f"Training for {steps_per_window} steps...")
model.learn(total_timesteps=steps_per_window, callback=eval_callback, progress_bar=True)
# Evaluate on Test Data
logger.info("Evaluating on unseen test window...")
model.set_env(eval_envs)
obs = eval_envs.reset()
dones = [False]
while not dones[0]:
action, _ = model.predict(obs, deterministic=True)
obs, rewards, dones, infos = eval_envs.step(action)
# Extract final metrics from info
info = infos[0] if isinstance(infos, list) else infos
if hasattr(eval_envs, 'envs'):
info = eval_envs.envs[0].get_episode_metrics()
elif isinstance(info, list):
info = info[0].get('episode_metrics', {})
logger.info(f"Test Window {i+1} Return: {info.get('total_return', 0)*100:.2f}% | Sharpe: {info.get('sharpe_ratio', 0):.2f}")
all_test_metrics.append(info)
# Save continuous model
model.save(self.model_dir / "continuous_agent.zip")
train_envs.save(str(self.model_dir / "continuous_agent_vecnorm.pkl"))
# Final Summary
avg_return = np.mean([m.get('total_return', 0) for m in all_test_metrics]) * 100
avg_sharpe = np.mean([m.get('sharpe_ratio', 0) for m in all_test_metrics])
logger.info("=" * 60)
logger.info("WALK-FORWARD CROSS VALIDATION COMPLETE")
logger.info(f"Average Out-of-Sample Return: {avg_return:.2f}%")
logger.info(f"Average Out-of-Sample Sharpe: {avg_sharpe:.2f}")
logger.info("=" * 60)
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--asset", type=str, default="BTCUSDT")
parser.add_argument("--days", type=int, default=730)
args = parser.parse_args()
trainer = WalkForwardTrainer(symbol=args.asset, total_days=args.days)
trainer.run()
|