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
90 lines
2.7 KiB
Python
90 lines
2.7 KiB
Python
"""Stockage des biais de marché.
|
|
|
|
- Redis : état COURANT (lu par la stratégie en live/dry-run). TTL court.
|
|
- Historique CSV horodaté par paire : trace durable pour backtester l'IA
|
|
(la stratégie en mode backtest lit le biais valide à chaque bougie).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import json
|
|
import os
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import redis
|
|
|
|
from models import MarketBias
|
|
|
|
_KEY_PREFIX = "bias:"
|
|
_DEFAULT_TTL = 3 * 3600 # 3 h : un biais expire s'il n'est pas rafraîchi
|
|
_HISTORY_HEADER = ["timestamp", "direction", "confidence", "key_support", "key_resistance"]
|
|
|
|
|
|
def _client() -> redis.Redis:
|
|
url = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
|
|
return redis.Redis.from_url(url, decode_responses=True)
|
|
|
|
|
|
def _key(pair: str) -> str:
|
|
return f"{_KEY_PREFIX}{pair}"
|
|
|
|
|
|
def write_bias(bias: MarketBias, ttl: int = _DEFAULT_TTL, r: Optional[redis.Redis] = None) -> None:
|
|
r = r or _client()
|
|
r.set(_key(bias.pair), bias.model_dump_json(), ex=ttl)
|
|
|
|
|
|
def read_bias(pair: str, r: Optional[redis.Redis] = None) -> Optional[MarketBias]:
|
|
r = r or _client()
|
|
raw = r.get(_key(pair))
|
|
if not raw:
|
|
return None
|
|
try:
|
|
return MarketBias.model_validate_json(raw)
|
|
except Exception: # noqa: BLE001 — donnée corrompue : on l'ignore
|
|
return None
|
|
|
|
|
|
def _history_path(history_dir: str, pair: str) -> Path:
|
|
return Path(history_dir) / f"{pair.replace('/', '_')}.csv"
|
|
|
|
|
|
def append_history(
|
|
bias: MarketBias, history_dir: str, ts: Optional[datetime] = None
|
|
) -> None:
|
|
"""Ajoute une ligne horodatée à l'historique CSV de la paire (créé si absent)."""
|
|
ts = ts or datetime.now(timezone.utc)
|
|
path = _history_path(history_dir, bias.pair)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
write_header = not path.exists()
|
|
with path.open("a", newline="", encoding="utf-8") as f:
|
|
writer = csv.writer(f)
|
|
if write_header:
|
|
writer.writerow(_HISTORY_HEADER)
|
|
writer.writerow(
|
|
[
|
|
ts.isoformat(),
|
|
bias.direction,
|
|
bias.confidence,
|
|
bias.key_support if bias.key_support is not None else "",
|
|
bias.key_resistance if bias.key_resistance is not None else "",
|
|
]
|
|
)
|
|
|
|
|
|
def read_all(r: Optional[redis.Redis] = None) -> dict[str, MarketBias]:
|
|
r = r or _client()
|
|
out: dict[str, MarketBias] = {}
|
|
for key in r.scan_iter(f"{_KEY_PREFIX}*"):
|
|
raw = r.get(key)
|
|
if not raw:
|
|
continue
|
|
try:
|
|
bias = MarketBias.model_validate_json(raw)
|
|
out[bias.pair] = bias
|
|
except Exception: # noqa: BLE001
|
|
continue
|
|
return out
|