341 lines
12 KiB
Python
341 lines
12 KiB
Python
"""Strategia 10: High Precision (target >80% accuracy).
|
|
Approccio: combina TUTTI gli indicatori disponibili, usa ensemble di 5 modelli,
|
|
trade SOLO quando tutti concordano. Pochi trade ma molto precisi.
|
|
Usa leva 3x per compensare bassa frequenza.
|
|
"""
|
|
from __future__ import annotations
|
|
import sys
|
|
sys.path.insert(0, ".")
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
from sklearn.ensemble import GradientBoostingClassifier, RandomForestClassifier, ExtraTreesClassifier
|
|
from sklearn.linear_model import LogisticRegression
|
|
from sklearn.svm import SVC
|
|
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
|
|
|
|
LEVERAGE = 3
|
|
FEE_PCT = 0.001
|
|
INITIAL_CAPITAL = 1000
|
|
|
|
|
|
def build_rich_features(df: pd.DataFrame, i: int) -> np.ndarray | None:
|
|
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 = []
|
|
|
|
for w in [6, 12, 24, 48, 96]:
|
|
if i < w:
|
|
feats.extend([0] * 18)
|
|
continue
|
|
|
|
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 = win_l.min(), max(win_h.max(), win_c.max())
|
|
rng = mx - mn if mx - mn > 0 else 1e-10
|
|
|
|
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)
|
|
|
|
feats.extend([
|
|
np.mean(rets) if len(rets) > 0 else 0,
|
|
np.std(rets) if len(rets) > 0 else 0,
|
|
np.sum(rets) if len(rets) > 0 else 0,
|
|
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),
|
|
np.mean(direction[-min(3, w):]),
|
|
(win_c[-1] - mn) / rng,
|
|
win_v[-1] / v_mean if v_mean > 0 else 1,
|
|
np.max(body) - np.min(body),
|
|
np.corrcoef(rets[:-1], rets[1:])[0, 1] if len(rets) > 1 and np.std(rets) > 0 else 0,
|
|
np.max(rets) if len(rets) > 0 else 0,
|
|
np.min(rets) if len(rets) > 0 else 0,
|
|
np.mean(np.abs(rets)) if len(rets) > 0 else 0,
|
|
np.sum(direction == 1) / w,
|
|
np.sum(direction == -1) / w,
|
|
])
|
|
|
|
# Hurst on different windows
|
|
for w in [48, 96]:
|
|
ret_w = np.diff(np.log(np.where(c[max(0,i-w):i] == 0, 1e-10, c[max(0,i-w):i])))
|
|
if len(ret_w) > 20:
|
|
feats.append(hurst_exponent(ret_w, max_lag=min(len(ret_w)//4, 15)))
|
|
else:
|
|
feats.append(0.5)
|
|
|
|
# Volatility ratios
|
|
feats.append(volatility_ratio(c[max(0,i-48):i], fast=6, slow=48))
|
|
feats.append(volatility_ratio(c[max(0,i-96):i], fast=12, slow=96))
|
|
|
|
# ATR normalized
|
|
tr = 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[1:])
|
|
feats.append(atr / c[i-1] if c[i-1] > 0 else 0)
|
|
|
|
# Position in range
|
|
h48 = np.max(h[i-48:i])
|
|
l48 = np.min(l[i-48:i])
|
|
r48 = h48 - l48
|
|
feats.append((c[i-1] - l48) / r48 if r48 > 0 else 0.5)
|
|
|
|
h96 = np.max(h[i-96:i])
|
|
l96 = np.min(l[i-96:i])
|
|
r96 = h96 - l96
|
|
feats.append((c[i-1] - l96) / r96 if r96 > 0 else 0.5)
|
|
|
|
return np.nan_to_num(np.array(feats), nan=0, posinf=1e6, neginf=-1e6)
|
|
|
|
|
|
def run_high_precision(asset: str, lookahead: int = 3):
|
|
print(f"\n{'#'*60}")
|
|
print(f" {asset} 1H — HIGH PRECISION (LA={lookahead})")
|
|
print(f"{'#'*60}")
|
|
|
|
df = load_data(asset, "1h")
|
|
close = df["close"].values
|
|
n = len(df)
|
|
|
|
MIN_RETURN = 0.003
|
|
|
|
# Build dataset
|
|
print(" Building features...")
|
|
X_all, y_all, idx_all = [], [], []
|
|
for i in range(200, n - lookahead, 1):
|
|
f = build_rich_features(df, i)
|
|
if f is None:
|
|
continue
|
|
ret = (close[i + lookahead - 1] - close[i - 1]) / close[i - 1]
|
|
if abs(ret) < MIN_RETURN:
|
|
continue
|
|
X_all.append(f)
|
|
y_all.append(1 if ret > 0 else 0)
|
|
idx_all.append(i)
|
|
|
|
X = np.array(X_all)
|
|
y = np.array(y_all)
|
|
idx_arr = np.array(idx_all)
|
|
|
|
print(f" Samples: {len(X)}, Features: {X.shape[1]}, Up: {np.mean(y)*100:.1f}%")
|
|
|
|
# Walk-forward with 5-model ensemble
|
|
TRAIN_SIZE = 15000
|
|
STEP_SIZE = 3000
|
|
|
|
models_config = [
|
|
("GB1", GradientBoostingClassifier(n_estimators=200, max_depth=4, min_samples_leaf=30, learning_rate=0.05, subsample=0.8, random_state=42)),
|
|
("GB2", GradientBoostingClassifier(n_estimators=300, max_depth=5, min_samples_leaf=50, learning_rate=0.03, subsample=0.7, random_state=123)),
|
|
("RF", RandomForestClassifier(n_estimators=300, max_depth=8, min_samples_leaf=30, class_weight="balanced", random_state=42, n_jobs=-1)),
|
|
("ET", ExtraTreesClassifier(n_estimators=300, max_depth=8, min_samples_leaf=30, class_weight="balanced", random_state=42, n_jobs=-1)),
|
|
("LR", LogisticRegression(max_iter=1000, C=0.1, random_state=42)),
|
|
]
|
|
|
|
capital = float(INITIAL_CAPITAL)
|
|
all_trades = []
|
|
equity = [capital]
|
|
|
|
fold = 0
|
|
start = 0
|
|
|
|
while start + TRAIN_SIZE + STEP_SIZE < len(X):
|
|
train_end = start + TRAIN_SIZE
|
|
test_end = min(train_end + STEP_SIZE, len(X))
|
|
|
|
X_tr, y_tr = X[start:train_end], y[start:train_end]
|
|
X_te, y_te = X[train_end:test_end], y[train_end:test_end]
|
|
idx_te = idx_arr[train_end:test_end]
|
|
|
|
scaler = StandardScaler()
|
|
X_tr_s = scaler.fit_transform(X_tr)
|
|
X_te_s = scaler.transform(X_te)
|
|
|
|
# Train all models
|
|
trained = []
|
|
for name, model in models_config:
|
|
m = type(model)(**model.get_params())
|
|
m.fit(X_tr_s, y_tr)
|
|
trained.append((name, m))
|
|
|
|
# Test with consensus voting
|
|
for j in range(len(X_te)):
|
|
votes_up = 0
|
|
votes_down = 0
|
|
max_conf = 0
|
|
|
|
for name, m in trained:
|
|
proba = m.predict_proba(X_te_s[j:j+1])[0]
|
|
up_idx = list(m.classes_).index(1)
|
|
p_up = proba[up_idx]
|
|
|
|
if p_up >= 0.60:
|
|
votes_up += 1
|
|
max_conf = max(max_conf, p_up)
|
|
elif p_up <= 0.40:
|
|
votes_down += 1
|
|
max_conf = max(max_conf, 1 - p_up)
|
|
|
|
i = idx_te[j]
|
|
actual_ret = (close[i + lookahead - 1] - close[i - 1]) / close[i - 1]
|
|
|
|
# Trade only with strong consensus
|
|
min_votes = 4 # at least 4 out of 5 models agree
|
|
direction = None
|
|
if votes_up >= min_votes:
|
|
direction = "long"
|
|
elif votes_down >= min_votes:
|
|
direction = "short"
|
|
|
|
if direction:
|
|
if direction == "long":
|
|
trade_ret = actual_ret
|
|
else:
|
|
trade_ret = -actual_ret
|
|
|
|
net_ret = trade_ret * LEVERAGE - FEE_PCT * 2 * LEVERAGE
|
|
pos_size = 0.2 # 20% of capital per trade
|
|
pnl = capital * pos_size * net_ret
|
|
capital += pnl
|
|
capital = max(capital, 0)
|
|
equity.append(capital)
|
|
|
|
is_correct = (direction == "long" and actual_ret > 0) or (direction == "short" and actual_ret < 0)
|
|
all_trades.append({
|
|
"fold": fold,
|
|
"idx": i,
|
|
"direction": direction,
|
|
"votes_up": votes_up,
|
|
"votes_down": votes_down,
|
|
"actual_ret": actual_ret,
|
|
"net_ret": net_ret,
|
|
"pnl": pnl,
|
|
"correct": is_correct,
|
|
})
|
|
|
|
fold += 1
|
|
start += STEP_SIZE
|
|
|
|
if not all_trades:
|
|
print(" No trades generated!")
|
|
return
|
|
|
|
trades_df = pd.DataFrame(all_trades)
|
|
n_correct = trades_df["correct"].sum()
|
|
n_total = len(trades_df)
|
|
accuracy = n_correct / n_total * 100
|
|
|
|
test_candles = idx_arr[-1] - idx_arr[TRAIN_SIZE]
|
|
test_days = test_candles / 24
|
|
test_years = test_days / 365.25
|
|
|
|
ann_ret = ((capital / INITIAL_CAPITAL) ** (1 / test_years) - 1) * 100 if test_years > 0 and capital > 0 else -100
|
|
daily_pnl = (capital - INITIAL_CAPITAL) / test_days if test_days > 0 else 0
|
|
|
|
# Max DD
|
|
peak = equity[0]
|
|
max_dd = 0
|
|
for v in equity:
|
|
if v > peak:
|
|
peak = v
|
|
dd = (peak - v) / peak if peak > 0 else 0
|
|
max_dd = max(max_dd, dd)
|
|
|
|
print(f"\n RISULTATI:")
|
|
print(f" Trades: {n_total}")
|
|
print(f" Accuracy: {accuracy:.1f}%")
|
|
print(f" Return: {(capital-INITIAL_CAPITAL)/INITIAL_CAPITAL*100:+.1f}%")
|
|
print(f" Annualized: {ann_ret:+.1f}%")
|
|
print(f" Max Drawdown: {max_dd*100:.1f}%")
|
|
print(f" Capital: €{capital:.0f}")
|
|
print(f" Trades/year: {n_total/test_years:.0f}")
|
|
print(f" €/day avg: €{daily_pnl:.2f}")
|
|
|
|
# Consensus threshold sweep
|
|
print(f"\n --- CONSENSUS SWEEP ---")
|
|
for min_v in [3, 4, 5]:
|
|
for ind_thr in [0.55, 0.60, 0.65]:
|
|
cap = float(INITIAL_CAPITAL)
|
|
trades_count = 0
|
|
correct_count = 0
|
|
eq = [cap]
|
|
|
|
fold_s = 0
|
|
start_s = 0
|
|
while start_s + TRAIN_SIZE + STEP_SIZE < len(X):
|
|
train_end_s = start_s + TRAIN_SIZE
|
|
test_end_s = min(train_end_s + STEP_SIZE, len(X))
|
|
|
|
X_tr_s2 = scaler.fit_transform(X[start_s:train_end_s])
|
|
X_te_s2 = scaler.transform(X[train_end_s:test_end_s])
|
|
y_tr_s2 = y[start_s:train_end_s]
|
|
idx_te_s2 = idx_arr[train_end_s:test_end_s]
|
|
|
|
trained_s = []
|
|
for name, model in models_config:
|
|
m2 = type(model)(**model.get_params())
|
|
m2.fit(X_tr_s2, y_tr_s2)
|
|
trained_s.append(m2)
|
|
|
|
for j in range(len(X_te_s2)):
|
|
vu = sum(1 for m2 in trained_s
|
|
if m2.predict_proba(X_te_s2[j:j+1])[0][list(m2.classes_).index(1)] >= ind_thr)
|
|
vd = sum(1 for m2 in trained_s
|
|
if m2.predict_proba(X_te_s2[j:j+1])[0][list(m2.classes_).index(1)] <= (1-ind_thr))
|
|
|
|
i_s = idx_te_s2[j]
|
|
ar = (close[i_s + lookahead - 1] - close[i_s - 1]) / close[i_s - 1]
|
|
|
|
d = None
|
|
if vu >= min_v:
|
|
d = "long"
|
|
elif vd >= min_v:
|
|
d = "short"
|
|
|
|
if d:
|
|
tr = ar if d == "long" else -ar
|
|
nr = tr * LEVERAGE - FEE_PCT * 2 * LEVERAGE
|
|
cap += cap * 0.2 * nr
|
|
cap = max(cap, 0)
|
|
eq.append(cap)
|
|
trades_count += 1
|
|
if (d == "long" and ar > 0) or (d == "short" and ar < 0):
|
|
correct_count += 1
|
|
|
|
start_s += STEP_SIZE
|
|
|
|
if trades_count > 0:
|
|
acc_s = correct_count / trades_count * 100
|
|
ret_s = (cap - INITIAL_CAPITAL) / INITIAL_CAPITAL * 100
|
|
ann_s = ((cap / INITIAL_CAPITAL) ** (1 / test_years) - 1) * 100 if test_years > 0 and cap > 0 else -100
|
|
dpnl = (cap - INITIAL_CAPITAL) / test_days if test_days > 0 else 0
|
|
print(f" min_votes={min_v} ind_thr={ind_thr:.2f}: trades={trades_count:4d} acc={acc_s:.1f}% ret={ret_s:+.1f}% ann={ann_s:+.1f}% €/day={dpnl:.2f}")
|
|
|
|
|
|
for asset in ["BTC", "ETH"]:
|
|
for la in [3, 6]:
|
|
run_high_precision(asset, la)
|