633b033f4d
- Infra: Freqtrade (futures dry-run) + Redis + dashboard + Docker Compose - Couche IA: ai_analyzer (Claude via abonnement, MCP TradingView, backfill biais) - Stratégies: SampleStrategy, AiBiasStrategy, IchimokuLS (long/short, validée train/test + données vierges + walk-forward), MTFIchimoku, variantes hyperopt - Arbitrage CEX (dry-run), backtesting, walk-forward, volatility targeting - IchimokuLS en dry-run live (config_live.json) Claude-Session: https://claude.ai/code/session_01VHETcFacdnDhQzthLpdYFR
93 lines
2.7 KiB
Python
93 lines
2.7 KiB
Python
"""Construction d'un instantané de marché compact à fournir à Claude.
|
|
|
|
On reste volontairement léger : prix courant, variation, et quelques statistiques
|
|
dérivées des dernières bougies (pas de dépendance lourde type talib). Claude raisonne
|
|
sur ce résumé ; le MCP TradingView (optionnel) peut enrichir l'analyse côté Claude.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, asdict
|
|
from typing import Optional
|
|
|
|
import ccxt
|
|
|
|
|
|
@dataclass
|
|
class PairSnapshot:
|
|
pair: str
|
|
timeframe: str
|
|
last_price: float
|
|
change_pct_window: float # variation % sur la fenêtre observée
|
|
high_window: float
|
|
low_window: float
|
|
sma_fast: float
|
|
sma_slow: float
|
|
n_candles: int
|
|
|
|
def to_dict(self) -> dict:
|
|
return asdict(self)
|
|
|
|
|
|
def _sma(values: list[float], period: int) -> Optional[float]:
|
|
if len(values) < period:
|
|
return None
|
|
return sum(values[-period:]) / period
|
|
|
|
|
|
def snapshot_from_candles(
|
|
pair: str,
|
|
timeframe: str,
|
|
ohlcv: list,
|
|
fast: int = 9,
|
|
slow: int = 21,
|
|
) -> Optional[PairSnapshot]:
|
|
"""Calcule un résumé à partir d'une liste de bougies OHLCV (live OU historique)."""
|
|
if not ohlcv:
|
|
return None
|
|
closes = [c[4] for c in ohlcv]
|
|
highs = [c[2] for c in ohlcv]
|
|
lows = [c[3] for c in ohlcv]
|
|
last = closes[-1]
|
|
first = closes[0]
|
|
return PairSnapshot(
|
|
pair=pair,
|
|
timeframe=timeframe,
|
|
last_price=round(last, 6),
|
|
change_pct_window=round((last - first) / first * 100, 2) if first else 0.0,
|
|
high_window=round(max(highs), 6),
|
|
low_window=round(min(lows), 6),
|
|
sma_fast=round(_sma(closes, fast) or last, 6),
|
|
sma_slow=round(_sma(closes, slow) or last, 6),
|
|
n_candles=len(closes),
|
|
)
|
|
|
|
|
|
def build_snapshot(
|
|
exchange_name: str,
|
|
pair: str,
|
|
timeframe: str = "1h",
|
|
limit: int = 100,
|
|
fast: int = 9,
|
|
slow: int = 21,
|
|
) -> Optional[PairSnapshot]:
|
|
"""Récupère les dernières bougies via ccxt et calcule un résumé."""
|
|
exchange_cls = getattr(ccxt, exchange_name)
|
|
exchange = exchange_cls({"enableRateLimit": True})
|
|
try:
|
|
ohlcv = exchange.fetch_ohlcv(pair, timeframe=timeframe, limit=limit)
|
|
except Exception as exc: # noqa: BLE001 — on journalise et on ignore la paire
|
|
print(f"[market_data] échec fetch {pair}: {exc}")
|
|
return None
|
|
return snapshot_from_candles(pair, timeframe, ohlcv, fast=fast, slow=slow)
|
|
|
|
|
|
def build_snapshots(
|
|
exchange_name: str, pairs: list[str], timeframe: str = "1h", limit: int = 100
|
|
) -> list[PairSnapshot]:
|
|
out = []
|
|
for pair in pairs:
|
|
snap = build_snapshot(exchange_name, pair, timeframe=timeframe, limit=limit)
|
|
if snap:
|
|
out.append(snap)
|
|
return out
|