PythagorasGoal/Old/scripts/waste/W06_structural_pattern.py

"""Strategia 6: Structural Pattern Matching con DTW veloce.
Idea diversa: invece di ML generico, cerca nel passato le finestre OHLC
più simili alla finestra corrente usando una versione veloce (reduced DTW).
Vota sulla direzione basandosi sui K-nearest neighbors nel passato.
Usa features normalizzate (non DTW puro sul prezzo che è lento).
"""
from __future__ import annotations
import sys
sys.path.insert(0, ".")

import numpy as np
import pandas as pd
from sklearn.neighbors import KNeighborsClassifier
from sklearn.preprocessing import StandardScaler
from src.data.downloader import load_data
from src.fractal.patterns import normalize_pattern_window

print("=" * 60)
print("  STRATEGIA 6: STRUCTURAL PATTERN KNN — BTC 1H")
print("=" * 60)

df = load_data("BTC", "1h")
close = df["close"].values
n = len(close)

WINDOW = 24
LOOKAHEAD = 6
MIN_RETURN = 0.003

def extract_structural_features(df: pd.DataFrame, idx: int, window: int) -> np.ndarray | None:
    """Extract normalized structural features from OHLC window."""
    if idx < window:
        return None

    o = df["open"].values[idx - window : idx]
    h = df["high"].values[idx - window : idx]
    l = df["low"].values[idx - window : idx]
    c = df["close"].values[idx - window : idx]
    v = df["volume"].values[idx - window : idx]

    # Normalize price to [0,1]
    all_prices = np.concatenate([o, h, l, c])
    mn, mx = all_prices.min(), all_prices.max()
    if mx - mn == 0:
        return None
    o_n = (o - mn) / (mx - mn)
    h_n = (h - mn) / (mx - mn)
    l_n = (l - mn) / (mx - mn)
    c_n = (c - mn) / (mx - mn)

    # Body and shadow ratios (already normalized)
    total = h - l
    total = np.where(total == 0, 1e-10, total)
    body = np.abs(c - o) / total
    upper_shadow = (h - np.maximum(o, c)) / total
    lower_shadow = (np.minimum(o, c) - l) / total
    direction = np.sign(c - o)

    # Returns
    log_c = np.log(np.where(c == 0, 1e-10, c))
    returns = np.diff(log_c)

    # Volume profile (normalized)
    v_mean = np.mean(v)
    v_n = v / v_mean if v_mean > 0 else np.ones_like(v)

    # Downsample to fixed-size feature vector
    # Take every N-th candle if window is large
    step = max(1, window // 12)
    sampled_idx = np.arange(0, window, step)[:12]

    features = np.concatenate([
        c_n[sampled_idx],           # 12: normalized close
        body[sampled_idx],          # 12: body ratios
        direction[sampled_idx],     # 12: direction
        upper_shadow[sampled_idx],  # 12: upper shadow
        lower_shadow[sampled_idx],  # 12: lower shadow
        v_n[sampled_idx],           # 12: volume profile
        [np.mean(returns), np.std(returns), np.sum(returns)],  # 3: return stats
        [np.mean(body), np.std(body)],  # 2: body stats
    ])
    return features


print("Extracting features...")
features_all = []
labels_all = []
indices_all = []

for i in range(WINDOW, n - LOOKAHEAD, 1):
    feats = extract_structural_features(df, i, WINDOW)
    if feats is None:
        continue

    future_ret = (close[i + LOOKAHEAD - 1] - close[i - 1]) / close[i - 1]
    if abs(future_ret) < MIN_RETURN:
        continue

    features_all.append(feats)
    labels_all.append(1 if future_ret > 0 else 0)
    indices_all.append(i)

X = np.array(features_all)
y = np.array(labels_all)
idx_arr = np.array(indices_all)

X = np.nan_to_num(X, nan=0.0, posinf=1e6, neginf=-1e6)

split = int(len(X) * 0.7)
X_train, X_test = X[:split], X[split:]
y_train, y_test = y[:split], y[split:]
idx_test = idx_arr[split:]

print(f"Samples: {len(X)} (train={split}, test={len(X)-split})")
print(f"Label balance: up={np.mean(y)*100:.1f}%")

scaler = StandardScaler()
X_train_s = scaler.fit_transform(X_train)
X_test_s = scaler.transform(X_test)

# Test diversi K
print("\n--- KNN SWEEP ---")
for K in [5, 10, 20, 50, 100, 200]:
    knn = KNeighborsClassifier(n_neighbors=K, weights="distance", n_jobs=-1)
    knn.fit(X_train_s, y_train)

    proba = knn.predict_proba(X_test_s)
    up_idx = list(knn.classes_).index(1)

    for thr in [0.55, 0.60, 0.65, 0.70]:
        sigs = []
        accs = []
        for j in range(len(X_test)):
            p_up = proba[j][up_idx]
            i = idx_test[j]
            actual = (close[i + LOOKAHEAD - 1] - close[i - 1]) / close[i - 1]

            if p_up >= thr:
                sigs.append(1)
                accs.append(1 if actual > 0 else 0)
            elif p_up <= (1 - thr):
                sigs.append(-1)
                accs.append(1 if actual < 0 else 0)

        if not accs:
            continue

        acc = np.mean(accs) * 100
        # PnL
        capital = 1000
        for direction, j in zip(sigs, range(len(accs))):
            i_idx = idx_test[[k for k in range(len(X_test)) if proba[k][up_idx] >= thr or proba[k][up_idx] <= (1 - thr)][j]]
            entry = close[i_idx - 1]
            exit_ = close[i_idx + LOOKAHEAD - 1]
            if direction == 1:
                ret = (exit_ - entry) / entry
            else:
                ret = (entry - exit_) / entry
            ret -= 0.002
            capital *= (1 + ret * 0.5)

        total_ret = (capital - 1000) / 1000 * 100
        print(f"  K={K:3d} thr={thr:.2f}: signals={len(accs):5d} acc={acc:.1f}% ret={total_ret:+.1f}%")

# Best combo: try with Gradient Boosting on same features
print("\n\n--- GRADIENT BOOSTING SU STRUCTURAL FEATURES ---")
from sklearn.ensemble import GradientBoostingClassifier

gb = GradientBoostingClassifier(
    n_estimators=300, max_depth=5, min_samples_leaf=30,
    learning_rate=0.03, subsample=0.8, random_state=42,
)
gb.fit(X_train_s, y_train)
proba_gb = gb.predict_proba(X_test_s)
up_idx_gb = list(gb.classes_).index(1)

for thr in [0.50, 0.55, 0.60, 0.65, 0.70, 0.75]:
    accs = []
    capital = 1000
    n_trades = 0
    for j in range(len(X_test)):
        p_up = proba_gb[j][up_idx_gb]
        i = idx_test[j]
        actual = (close[i + LOOKAHEAD - 1] - close[i - 1]) / close[i - 1]

        if p_up >= thr:
            accs.append(1 if actual > 0 else 0)
            ret = actual - 0.002
            capital *= (1 + ret * 0.5)
            n_trades += 1
        elif p_up <= (1 - thr):
            accs.append(1 if actual < 0 else 0)
            ret = -actual - 0.002
            capital *= (1 + ret * 0.5)
            n_trades += 1

    if not accs:
        continue
    acc = np.mean(accs) * 100
    total_ret = (capital - 1000) / 1000 * 100
    print(f"  thr={thr:.2f}: trades={n_trades:5d} acc={acc:.1f}% ret={total_ret:+.1f}%")