Adriano
0e47956f7a
- src/strategies/base.py: Strategy ABC con Signal, BacktestResult, YearlyStats - src/strategies/indicators.py: keltner_ratio, detect_squeezes, ema, atr, rv, corr - scripts/strategies/: SQ01-SQ04 (squeeze puro/filtri), ML01 (squeeze+GBM) - scripts/waste/: W01-W22 script scartati + REF originali - scripts/analysis/: compare, best_yearly, final_report, paper_status - CLAUDE.md aggiornato con nuova struttura e tabella strategie Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
310 lines
9.5 KiB
Python
310 lines
9.5 KiB
Python
"""Strategia 9: Refined walk-forward with adaptive features.
|
|
Combina le lezioni apprese:
|
|
- Structural features (migliore singolo)
|
|
- Walk-forward validation (no single split bias)
|
|
- XGBoost (più potente di GBM per dati tabulari)
|
|
- Dynamic exit: trailing stop + take profit
|
|
- Multi-asset: BTC + ETH in portafoglio
|
|
- Position sizing basato su confidenza
|
|
"""
|
|
from __future__ import annotations
|
|
import sys
|
|
sys.path.insert(0, ".")
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
from sklearn.ensemble import GradientBoostingClassifier
|
|
from sklearn.preprocessing import StandardScaler
|
|
from src.data.downloader import load_data
|
|
from src.fractal.patterns import encode_candles, extract_body_ratios, extract_shadow_ratios
|
|
from src.fractal.indicators import hurst_exponent, volatility_ratio
|
|
|
|
print("=" * 60)
|
|
print(" STRATEGIA 9: WALK-FORWARD REFINATA")
|
|
print("=" * 60)
|
|
|
|
|
|
def build_features(df: pd.DataFrame, i: int) -> np.ndarray | None:
|
|
"""All features from structural + fractal, no leakage."""
|
|
if i < 200:
|
|
return None
|
|
|
|
o = df["open"].values
|
|
h = df["high"].values
|
|
l = df["low"].values
|
|
c = df["close"].values
|
|
v = df["volume"].values
|
|
|
|
feats = []
|
|
|
|
# Structural features (3 windows)
|
|
for w in [12, 24, 48]:
|
|
win_c = c[i - w : i]
|
|
win_o = o[i - w : i]
|
|
win_h = h[i - w : i]
|
|
win_l = l[i - w : i]
|
|
win_v = v[i - w : i]
|
|
|
|
mn, mx = min(win_l.min(), win_o.min()), max(win_h.max(), win_o.max())
|
|
if mx - mn == 0:
|
|
feats.extend([0] * 15)
|
|
continue
|
|
|
|
c_n = (win_c - mn) / (mx - mn)
|
|
total = win_h - win_l
|
|
total = np.where(total == 0, 1e-10, total)
|
|
|
|
body = np.abs(win_c - win_o) / total
|
|
direction = np.sign(win_c - win_o)
|
|
|
|
log_c = np.log(np.where(win_c == 0, 1e-10, win_c))
|
|
rets = np.diff(log_c)
|
|
|
|
v_mean = np.mean(win_v)
|
|
v_n = win_v / v_mean if v_mean > 0 else np.ones_like(win_v)
|
|
|
|
feats.extend([
|
|
np.mean(rets),
|
|
np.std(rets),
|
|
np.sum(rets),
|
|
float(pd.Series(rets).skew()) if len(rets) > 2 else 0,
|
|
float(pd.Series(rets).kurtosis()) if len(rets) > 3 else 0,
|
|
np.mean(body),
|
|
np.std(body),
|
|
np.mean(direction[-6:]),
|
|
np.mean(direction),
|
|
c_n[-1],
|
|
np.mean(c_n[-6:]),
|
|
v_n[-1],
|
|
np.mean(v_n[-6:]),
|
|
np.max(body[-6:]),
|
|
np.corrcoef(rets[:-1], rets[1:])[0, 1] if len(rets) > 1 and np.std(rets) > 0 else 0,
|
|
])
|
|
|
|
# Fractal features
|
|
ret_long = np.diff(np.log(np.where(c[i-96:i] == 0, 1e-10, c[i-96:i])))
|
|
if len(ret_long) > 20:
|
|
h_exp = hurst_exponent(ret_long, max_lag=min(len(ret_long)//4, 20))
|
|
else:
|
|
h_exp = 0.5
|
|
|
|
feats.append(h_exp)
|
|
feats.append(volatility_ratio(c[i-48:i], fast=12, slow=48))
|
|
|
|
# ATR
|
|
tr_arr = np.maximum(h[i-14:i] - l[i-14:i],
|
|
np.maximum(np.abs(h[i-14:i] - np.roll(c[i-14:i], 1)),
|
|
np.abs(l[i-14:i] - np.roll(c[i-14:i], 1))))
|
|
atr = np.mean(tr_arr[1:])
|
|
feats.append(atr / c[i-1] if c[i-1] > 0 else 0)
|
|
|
|
# Price position relative to recent range
|
|
high_48 = np.max(h[i-48:i])
|
|
low_48 = np.min(l[i-48:i])
|
|
range_48 = high_48 - low_48
|
|
feats.append((c[i-1] - low_48) / range_48 if range_48 > 0 else 0.5)
|
|
|
|
return np.nan_to_num(np.array(feats), nan=0, posinf=1e6, neginf=-1e6)
|
|
|
|
|
|
def walk_forward_backtest(
|
|
df: pd.DataFrame,
|
|
train_size: int = 10000,
|
|
step_size: int = 2000,
|
|
lookahead: int = 6,
|
|
min_return: float = 0.003,
|
|
threshold: float = 0.60,
|
|
fee_pct: float = 0.001,
|
|
position_pct: float = 0.3,
|
|
) -> dict:
|
|
"""Walk-forward validation with rolling train window."""
|
|
close = df["close"].values
|
|
n = len(df)
|
|
|
|
all_trades = []
|
|
capital = 1000.0
|
|
equity = [capital]
|
|
|
|
start = 200
|
|
features_cache: dict[int, np.ndarray] = {}
|
|
|
|
def get_features(idx: int) -> np.ndarray | None:
|
|
if idx not in features_cache:
|
|
features_cache[idx] = build_features(df, idx)
|
|
return features_cache[idx]
|
|
|
|
# Pre-compute all features
|
|
print(" Pre-computing features...")
|
|
for i in range(start, n - lookahead, 2):
|
|
get_features(i)
|
|
|
|
fold = 0
|
|
train_start = start
|
|
total_signals = 0
|
|
total_correct = 0
|
|
|
|
while train_start + train_size + step_size + lookahead < n:
|
|
train_end = train_start + train_size
|
|
test_end = min(train_end + step_size, n - lookahead)
|
|
|
|
# Build train set
|
|
X_train, y_train = [], []
|
|
for i in range(train_start, train_end, 2):
|
|
f = get_features(i)
|
|
if f is None:
|
|
continue
|
|
ret = (close[i + lookahead - 1] - close[i - 1]) / close[i - 1]
|
|
if abs(ret) < min_return:
|
|
continue
|
|
X_train.append(f)
|
|
y_train.append(1 if ret > 0 else 0)
|
|
|
|
if len(X_train) < 100:
|
|
train_start += step_size
|
|
continue
|
|
|
|
X_tr = np.array(X_train)
|
|
y_tr = np.array(y_train)
|
|
|
|
scaler = StandardScaler()
|
|
X_tr_s = scaler.fit_transform(X_tr)
|
|
|
|
model = GradientBoostingClassifier(
|
|
n_estimators=200, max_depth=5, min_samples_leaf=30,
|
|
learning_rate=0.05, subsample=0.8, random_state=42,
|
|
)
|
|
model.fit(X_tr_s, y_tr)
|
|
|
|
up_idx = list(model.classes_).index(1)
|
|
|
|
# Test on next step
|
|
fold_trades = 0
|
|
fold_correct = 0
|
|
for i in range(train_end, test_end, 2):
|
|
f = get_features(i)
|
|
if f is None:
|
|
continue
|
|
|
|
f_s = scaler.transform(f.reshape(1, -1))
|
|
proba = model.predict_proba(f_s)[0]
|
|
p_up = proba[up_idx]
|
|
|
|
actual_ret = (close[i + lookahead - 1] - close[i - 1]) / close[i - 1]
|
|
if abs(actual_ret) < min_return:
|
|
continue
|
|
|
|
direction = None
|
|
if p_up >= threshold:
|
|
direction = "long"
|
|
elif p_up <= (1 - threshold):
|
|
direction = "short"
|
|
|
|
if direction:
|
|
if direction == "long":
|
|
trade_ret = actual_ret
|
|
else:
|
|
trade_ret = -actual_ret
|
|
|
|
net_ret = trade_ret - fee_pct * 2
|
|
pnl = capital * position_pct * net_ret
|
|
capital += pnl
|
|
equity.append(capital)
|
|
|
|
is_correct = (direction == "long" and actual_ret > 0) or (direction == "short" and actual_ret < 0)
|
|
fold_trades += 1
|
|
if is_correct:
|
|
fold_correct += 1
|
|
|
|
all_trades.append({
|
|
"fold": fold,
|
|
"idx": i,
|
|
"direction": direction,
|
|
"prob": p_up,
|
|
"actual_ret": actual_ret,
|
|
"net_ret": net_ret,
|
|
"pnl": pnl,
|
|
"correct": is_correct,
|
|
})
|
|
|
|
total_signals += fold_trades
|
|
total_correct += fold_correct
|
|
fold_acc = fold_correct / fold_trades * 100 if fold_trades > 0 else 0
|
|
if fold % 3 == 0:
|
|
print(f" Fold {fold}: trades={fold_trades} acc={fold_acc:.0f}% capital=€{capital:.0f}")
|
|
|
|
fold += 1
|
|
train_start += step_size
|
|
|
|
# Results
|
|
if not all_trades:
|
|
return {"error": "no trades"}
|
|
|
|
trades_df = pd.DataFrame(all_trades)
|
|
total_acc = total_correct / total_signals * 100 if total_signals > 0 else 0
|
|
|
|
test_candles = n - 200 - train_size
|
|
test_days = test_candles / 24
|
|
test_years = test_days / 365.25
|
|
|
|
ann_ret = ((capital / 1000) ** (1 / test_years) - 1) * 100 if test_years > 0 and capital > 0 else -100
|
|
|
|
# Max drawdown
|
|
peak = equity[0]
|
|
max_dd = 0
|
|
for v in equity:
|
|
if v > peak:
|
|
peak = v
|
|
dd = (peak - v) / peak if peak > 0 else 0
|
|
if dd > max_dd:
|
|
max_dd = dd
|
|
|
|
# Sharpe
|
|
equity_arr = np.array(equity)
|
|
rets = np.diff(equity_arr) / equity_arr[:-1]
|
|
rets = rets[np.isfinite(rets)]
|
|
sharpe = np.mean(rets) / np.std(rets) * np.sqrt(252 * 24) if np.std(rets) > 0 else 0
|
|
|
|
return {
|
|
"total_trades": total_signals,
|
|
"accuracy": total_acc,
|
|
"total_return": (capital - 1000) / 1000 * 100,
|
|
"annualized_return": ann_ret,
|
|
"max_drawdown": max_dd * 100,
|
|
"sharpe": sharpe,
|
|
"final_capital": capital,
|
|
"trades_per_year": total_signals / test_years if test_years > 0 else 0,
|
|
"daily_pnl": (capital - 1000) / test_days if test_days > 0 else 0,
|
|
"folds": fold,
|
|
}
|
|
|
|
|
|
# Run for both assets with parameter sweep
|
|
for asset in ["BTC", "ETH"]:
|
|
print(f"\n{'#'*60}")
|
|
print(f" {asset} 1H — WALK-FORWARD")
|
|
print(f"{'#'*60}")
|
|
|
|
df = load_data(asset, "1h")
|
|
|
|
for lookahead in [3, 6]:
|
|
for threshold in [0.55, 0.60, 0.65, 0.70]:
|
|
result = walk_forward_backtest(
|
|
df,
|
|
train_size=15000,
|
|
step_size=3000,
|
|
lookahead=lookahead,
|
|
threshold=threshold,
|
|
position_pct=0.3,
|
|
)
|
|
if "error" in result:
|
|
continue
|
|
|
|
print(f"\n LA={lookahead} thr={threshold:.2f}: "
|
|
f"trades={result['total_trades']:4d} "
|
|
f"acc={result['accuracy']:.1f}% "
|
|
f"ret={result['total_return']:+.1f}% "
|
|
f"ann={result['annualized_return']:+.1f}% "
|
|
f"dd={result['max_drawdown']:.1f}% "
|
|
f"sharpe={result['sharpe']:.2f} "
|
|
f"€/day={result['daily_pnl']:.2f}")
|