AKQuant provides a structured workflow for Machine Learning strategies using akquant.ml.QuantModel and adapters like SklearnAdapter or PyTorchAdapter.
Workflow:
- Initialization: In
__init__, initialize self.model with an adapter. - Configuration: Use
self.model.set_validation(...) to configure Walk-Forward Validation. This automates rolling windows and training triggers. - Feature Engineering: Implement
prepare_features(self, df, mode):mode='training': Return (X, y). Ensure y (e.g., shifted returns) is aligned with X and drop NaNs.mode='inference': Return X (the features for the current bar).
- Training: The framework automatically triggers
on_train_signal $\rightarrow$ prepare_features(mode='training') $\rightarrow$ model.fit(). - Inference: In
on_bar, check self.is_model_ready() and self.current_validation_window(), then call prepare_features(mode='inference') and model.predict(). - Lifecycle: Training occurs on the current bar, but the model activates on the next bar. The framework calls
model.clone() for each training window.
Key Validation Parameters for set_validation:
method: e.g., 'walk_forward'.train_window: Duration for training (e.g., '200d').test_window: Planned Out-of-Sample (OOS) range (e.g., '30d').rolling_step: Frequency of retraining (e.g., '30d').
from akquant import Strategy, Bar
from akquant.ml import SklearnAdapter
from sklearn.ensemble import RandomForestClassifier
import pandas as pd
import numpy as np
class MLStrategy(Strategy):
def __init__(self):
# 1. Initialize Adapter
self.model = SklearnAdapter(RandomForestClassifier(n_estimators=10))
# 2. Configure Walk-Forward (Auto-Training)
self.model.set_validation(
method='walk_forward',
train_window='200d',
test_window='30d',
rolling_step='30d',
frequency='1d',
verbose=True
)
def prepare_features(self, df: pd.DataFrame, mode: str = "training"):
df['ret1'] = df['close'].pct_change()
df['ret5'] = df['close'].pct_change(5)
df['vol_change'] = df['volume'].pct_change()
features = ['ret1', 'ret5', 'vol_change']
if mode == 'inference':
return df[features].iloc[-1:].fillna(0)
df['target'] = (df['close'].shift(-1) > df['close']).astype(int)
data = df.dropna()
return data[features], data['target']
def on_bar(self, bar: Bar):
window = self.current_validation_window()
if window is None or not self.is_model_ready():
return
hist_df = self.get_history_df(30)
if len(hist_df) < 10:
return
X_curr = self.prepare_features(hist_df, mode='inference')
try:
pred = self.model.predict(X_curr)[0]
pos = self.get_position(bar.symbol)
if pred == 1 and pos == 0:
self.buy(bar.symbol, 1000)
elif pred == 0 and pos > 0:
self.sell(bar.symbol, pos)
except Exception:
pass
