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>
111 lines
3.9 KiB
Python
111 lines
3.9 KiB
Python
"""Analisi baseline: distribuzione pattern frattali e prima strategia naive."""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
sys.path.insert(0, ".")
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
from src.data.downloader import load_data
|
|
from src.fractal.patterns import encode_candles, find_patterns, pattern_frequency
|
|
from src.backtest.engine import run_backtest, BacktestResult
|
|
|
|
print("=" * 60)
|
|
print(" ANALISI BASELINE — BTC 1H")
|
|
print("=" * 60)
|
|
|
|
df = load_data("BTC", "1h")
|
|
print(f"\nDati: {len(df)} candele [{df['datetime'].iloc[0]} → {df['datetime'].iloc[-1]}]")
|
|
|
|
# 1. Distribuzione pattern
|
|
print("\n--- DISTRIBUZIONE PATTERN (3-6 candele) ---")
|
|
candle_types = encode_candles(df)
|
|
unique, counts = np.unique(candle_types, return_counts=True)
|
|
type_map = {-1: "DOWN", 0: "DOJI", 1: "UP"}
|
|
for t, c in zip(unique, counts):
|
|
print(f" {type_map[t]}: {c} ({c/len(df)*100:.1f}%)")
|
|
|
|
patterns = find_patterns(df, min_len=3, max_len=6)
|
|
freq = pattern_frequency(patterns)
|
|
print(f"\nPattern unici: {len(freq)}")
|
|
print(f"\nTop 20 pattern più frequenti:")
|
|
print(freq.head(20).to_string(index=False))
|
|
|
|
# 2. Analisi predittiva: dopo ogni pattern, il prezzo sale o scende?
|
|
print("\n\n--- ANALISI PREDITTIVA PER PATTERN ---")
|
|
print("Per ogni pattern: % volte che il prezzo sale nelle N candele successive")
|
|
|
|
LOOKAHEAD = [1, 3, 6, 12, 24]
|
|
top_patterns = freq.head(30)["pattern"].tolist()
|
|
|
|
results = []
|
|
for code in top_patterns:
|
|
matching = [p for p in patterns if p.code == code]
|
|
if len(matching) < 50:
|
|
continue
|
|
|
|
row = {"pattern": code, "count": len(matching)}
|
|
for ahead in LOOKAHEAD:
|
|
ups = 0
|
|
valid = 0
|
|
for p in matching:
|
|
future_idx = p.end_idx + ahead
|
|
if future_idx >= len(df):
|
|
continue
|
|
valid += 1
|
|
if df["close"].iloc[future_idx] > df["close"].iloc[p.end_idx - 1]:
|
|
ups += 1
|
|
if valid > 0:
|
|
row[f"up_{ahead}h"] = round(ups / valid * 100, 1)
|
|
else:
|
|
row[f"up_{ahead}h"] = None
|
|
results.append(row)
|
|
|
|
pred_df = pd.DataFrame(results)
|
|
print(pred_df.to_string(index=False))
|
|
|
|
# 3. Strategia naive: compra quando il pattern più bullish si presenta
|
|
print("\n\n--- STRATEGIA 1: PATTERN BULLISH NAIVE ---")
|
|
# Trova pattern con up_24h > 55%
|
|
bullish_patterns = pred_df[pred_df["up_24h"] > 55]["pattern"].tolist()
|
|
bearish_patterns = pred_df[pred_df["up_24h"] < 45]["pattern"].tolist()
|
|
print(f"Pattern bullish (>55% up in 24h): {len(bullish_patterns)}")
|
|
print(f"Pattern bearish (<45% up in 24h): {len(bearish_patterns)}")
|
|
|
|
# Genera segnali
|
|
signals = pd.Series(0, index=df.index)
|
|
all_patterns = find_patterns(df, min_len=3, max_len=6)
|
|
for p in all_patterns:
|
|
if p.code in bullish_patterns:
|
|
signals.iloc[p.end_idx - 1] = 1
|
|
elif p.code in bearish_patterns:
|
|
if signals.iloc[p.end_idx - 1] == 0:
|
|
signals.iloc[p.end_idx - 1] = -1
|
|
|
|
# Train/test split: 70/30
|
|
split_idx = int(len(df) * 0.7)
|
|
train_df = df.iloc[:split_idx].reset_index(drop=True)
|
|
test_df = df.iloc[split_idx:].reset_index(drop=True)
|
|
train_signals = signals.iloc[:split_idx].reset_index(drop=True)
|
|
test_signals = signals.iloc[split_idx:].reset_index(drop=True)
|
|
|
|
train_result = run_backtest(train_df, train_signals, initial_capital=1000, fee_pct=0.001)
|
|
test_result = run_backtest(test_df, test_signals, initial_capital=1000, fee_pct=0.001)
|
|
|
|
print("\nRISULTATI TRAIN (70%):")
|
|
for k, v in train_result.summary().items():
|
|
print(f" {k}: {v}")
|
|
|
|
print("\nRISULTATI TEST (30%):")
|
|
for k, v in test_result.summary().items():
|
|
print(f" {k}: {v}")
|
|
|
|
# 4. Buy & Hold come benchmark
|
|
print("\n\n--- BENCHMARK: BUY & HOLD ---")
|
|
bh_signals = pd.Series(0, index=test_df.index)
|
|
bh_signals.iloc[0] = 1 # Compra al primo candle
|
|
bh_result = run_backtest(test_df, bh_signals, initial_capital=1000, fee_pct=0.001, max_hold_candles=len(test_df))
|
|
print("Buy & Hold (test period):")
|
|
for k, v in bh_result.summary().items():
|
|
print(f" {k}: {v}")
|