MidasBot: bot trading crypto IA + stratégies Ichimoku validées

- 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
This commit is contained in:
jerem
2026-06-23 19:25:49 +02:00
commit 633b033f4d
59 changed files with 3868 additions and 0 deletions

23
ai_analyzer/Dockerfile Normal file
View File

@@ -0,0 +1,23 @@
# Service ai-analyzer : Python + CLI Claude (Node), authentifié par l'abonnement.
FROM node:20-slim
RUN apt-get update \
&& apt-get install -y --no-install-recommends python3 python3-pip \
&& rm -rf /var/lib/apt/lists/*
# CLI Claude Code (mode headless via `claude -p`).
RUN npm install -g @anthropic-ai/claude-code
WORKDIR /app
COPY ai_analyzer/requirements.txt .
RUN pip3 install --no-cache-dir --break-system-packages -r requirements.txt
COPY ai_analyzer/ /app/
COPY .mcp.json /app/.mcp.json
ENV REDIS_URL=redis://redis:6379/0 \
ANALYZER_MODEL=claude-sonnet-4-6
# Auth : CLAUDE_CODE_OAUTH_TOKEN fourni par l'environnement (cf. .env).
# NE PAS définir ANTHROPIC_API_KEY (basculerait sur la facturation au token).
CMD ["python3", "analyzer.py"]

92
ai_analyzer/analyzer.py Normal file
View File

@@ -0,0 +1,92 @@
"""Boucle d'analyse : marché -> Claude -> biais dans Redis.
Cadence VOLONTAIREMENT basse (1 cycle par bougie sur timeframe >= 15 min) :
l'usage automatisé d'un abonnement a des limites ; un cycle = UN seul appel Claude
couvrant toutes les paires (batch), pour économiser le quota.
Usage :
python analyzer.py --once # un seul cycle (test)
python analyzer.py # boucle continue
"""
from __future__ import annotations
import argparse
import os
import time
from datetime import datetime, timezone
from pathlib import Path
from claude_client import ClaudeClient, ClaudeError
from market_data import build_snapshots
from signal_store import append_history, write_bias
_PROJECT_ROOT = Path(__file__).resolve().parent.parent
# --- Configuration via variables d'environnement ---
EXCHANGE = os.environ.get("ANALYZER_EXCHANGE", "binance")
PAIRS = os.environ.get("ANALYZER_PAIRS", "BTC/USDT,ETH/USDT,SOL/USDT,BNB/USDT").split(",")
TIMEFRAME = os.environ.get("ANALYZER_TIMEFRAME", "1h")
INTERVAL_S = int(os.environ.get("ANALYZER_INTERVAL_S", "3600")) # 1 h par défaut
MODEL = os.environ.get("ANALYZER_MODEL", "claude-sonnet-4-6")
MCP_CONFIG = os.environ.get("ANALYZER_MCP_CONFIG") # ex. "/app/.mcp.json"
ALLOWED_TOOLS = os.environ.get("ANALYZER_ALLOWED_TOOLS") # ex. "mcp__tradingview"
# Historique horodaté (pour backtester l'IA). Lu par AiBiasStrategy en backtest.
HISTORY_DIR = os.environ.get(
"ANALYZER_HISTORY_DIR",
str(_PROJECT_ROOT / "freqtrade" / "user_data" / "ai_bias_history"),
)
def _log(msg: str) -> None:
ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
print(f"[analyzer {ts}] {msg}", flush=True)
def run_cycle(client: ClaudeClient) -> int:
pairs = [p.strip() for p in PAIRS if p.strip()]
_log(f"Instantané {EXCHANGE} {TIMEFRAME} pour {pairs}")
snapshots = build_snapshots(EXCHANGE, pairs, timeframe=TIMEFRAME)
if not snapshots:
_log("Aucun instantané récupéré — cycle ignoré.")
return 0
_log(f"Appel Claude ({MODEL}) pour {len(snapshots)} paire(s)…")
batch = client.get_biases(snapshots)
written = 0
ts = datetime.now(timezone.utc)
for bias in batch.biases:
write_bias(bias) # état courant (live/dry-run)
append_history(bias, HISTORY_DIR, ts) # trace horodatée (backtest IA)
written += 1
_log(
f"{bias.pair}: {bias.direction} (conf={bias.confidence:.2f}) — {bias.rationale}"
)
_log(f"{written} biais écrit(s) dans Redis + historique ({HISTORY_DIR}).")
return written
def main() -> None:
parser = argparse.ArgumentParser(description="MidasBot — analyzer IA")
parser.add_argument("--once", action="store_true", help="un seul cycle puis sortie")
args = parser.parse_args()
client = ClaudeClient(model=MODEL, mcp_config=MCP_CONFIG, allowed_tools=ALLOWED_TOOLS)
if args.once:
run_cycle(client)
return
_log(f"Démarrage boucle (intervalle {INTERVAL_S}s).")
while True:
try:
run_cycle(client)
except ClaudeError as exc:
_log(f"Erreur Claude : {exc}")
except Exception as exc: # noqa: BLE001
_log(f"Erreur inattendue : {exc}")
time.sleep(INTERVAL_S)
if __name__ == "__main__":
main()

111
ai_analyzer/backfill.py Normal file
View File

@@ -0,0 +1,111 @@
"""Backfill — génère un historique de biais IA pour backtester la stratégie.
Parcourt l'historique des bougies à une cadence donnée et, à chaque pas de temps,
demande à Claude un biais en n'utilisant QUE les données disponibles jusqu'à ce
moment-là (pas de fuite du futur). Écrit chaque biais dans l'historique CSV avec
l'horodatage de la bougie correspondante.
⚠️ COÛTEUX EN QUOTA D'ABONNEMENT : 1 appel Claude par pas de temps.
6 mois × cadence 24 h ≈ 180 appels. Choisis une cadence raisonnable.
Usage :
python backfill.py --start 20260101 --step-hours 24
python backfill.py --start 20260101 --end 20260301 --step-hours 12 --pairs BTC/USDT,ETH/USDT
"""
from __future__ import annotations
import argparse
import os
from datetime import datetime, timedelta, timezone
import ccxt
from claude_client import ClaudeClient
from market_data import snapshot_from_candles
from signal_store import append_history
EXCHANGE = os.environ.get("ANALYZER_EXCHANGE", "binance")
MODEL = os.environ.get("ANALYZER_MODEL", "claude-sonnet-4-6")
HISTORY_DIR = os.environ.get(
"ANALYZER_HISTORY_DIR",
os.path.join(os.path.dirname(os.path.dirname(__file__)),
"freqtrade", "user_data", "ai_bias_history"),
)
WINDOW = 100 # nb de bougies fournies à Claude à chaque pas
def _parse_day(s: str) -> datetime:
return datetime.strptime(s, "%Y%m%d").replace(tzinfo=timezone.utc)
def _fetch_full(exchange, pair: str, timeframe: str, since_ms: int) -> list:
"""Récupère tout l'OHLCV depuis `since_ms` (pagination ccxt)."""
out: list = []
cursor = since_ms
while True:
batch = exchange.fetch_ohlcv(pair, timeframe=timeframe, since=cursor, limit=1000)
if not batch:
break
out += batch
cursor = batch[-1][0] + 1
if len(batch) < 1000:
break
return out
def main() -> None:
p = argparse.ArgumentParser(description="MidasBot — backfill historique des biais IA")
p.add_argument("--pairs", default="BTC/USDT,ETH/USDT,SOL/USDT,BNB/USDT")
p.add_argument("--timeframe", default="1h")
p.add_argument("--start", required=True, help="AAAAMMJJ")
p.add_argument("--end", default=None, help="AAAAMMJJ (défaut: maintenant)")
p.add_argument("--step-hours", type=int, default=24, help="cadence d'analyse")
p.add_argument("--yes", action="store_true", help="ne pas demander confirmation")
args = p.parse_args()
pairs = [x.strip() for x in args.pairs.split(",") if x.strip()]
tf = args.timeframe
start = _parse_day(args.start)
end = _parse_day(args.end) if args.end else datetime.now(timezone.utc)
step = timedelta(hours=args.step_hours)
n_steps = int((end - start) / step)
print(f"Backfill {pairs} {tf} | {start.date()}{end.date()} | "
f"pas {args.step_hours} h | ~{n_steps} appels Claude ({MODEL})")
if not args.yes:
if input("Continuer ? Ça consomme ton quota d'abonnement [y/N] ").strip().lower() != "y":
print("Annulé.")
return
exchange = getattr(ccxt, EXCHANGE)({"enableRateLimit": True})
since_ms = int(start.timestamp() * 1000)
# Récupère tout l'historique une fois par paire (puis on tranche par pas de temps).
full = {pair: _fetch_full(exchange, pair, tf, since_ms) for pair in pairs}
client = ClaudeClient(model=MODEL)
t = start
done = 0
while t <= end:
t_ms = int(t.timestamp() * 1000)
snaps = []
for pair in pairs:
candles = [c for c in full[pair] if c[0] <= t_ms][-WINDOW:]
snap = snapshot_from_candles(pair, tf, candles)
if snap:
snaps.append(snap)
if snaps:
try:
batch = client.get_biases(snaps)
for bias in batch.biases:
append_history(bias, HISTORY_DIR, ts=t)
done += 1
print(f" {t.isoformat()}{len(batch.biases)} biais")
except Exception as exc: # noqa: BLE001
print(f" {t.isoformat()} → erreur: {exc}")
t += step
print(f"Terminé : {done} pas écrits dans {HISTORY_DIR}")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,154 @@
"""Client Claude headless via le CLI `claude` (authentifié par l'abonnement).
On n'utilise PAS le SDK `anthropic` ni de clé API. On invoque `claude -p` en
sous-processus avec sortie structurée (`--json-schema`), ce qui consomme l'abonnement
Claude Code de l'utilisateur (pas de facturation au token).
"""
from __future__ import annotations
import json
import os
import subprocess
from pathlib import Path
from typing import Optional
from market_data import PairSnapshot
from models import MarketBias, MarketBiasBatch
_HERE = Path(__file__).resolve().parent
_SYSTEM_PROMPT_FILE = _HERE / "system_prompt.md"
class ClaudeError(RuntimeError):
pass
class ClaudeClient:
def __init__(
self,
model: str = "claude-sonnet-4-6",
max_turns: int = 6,
mcp_config: Optional[str] = None,
allowed_tools: Optional[str] = None,
timeout_s: int = 180,
) -> None:
self.model = model
self.max_turns = max_turns
# .mcp.json à la racine projet, chargé seulement si présent.
self.mcp_config = mcp_config
# Ex. "mcp__tradingview" pour autoriser uniquement les outils du serveur MCP.
self.allowed_tools = allowed_tools
self.timeout_s = timeout_s
self._guard_no_api_key()
@staticmethod
def _guard_no_api_key() -> None:
# ANTHROPIC_API_KEY ferait basculer Claude sur la facturation au token.
if os.environ.get("ANTHROPIC_API_KEY"):
raise ClaudeError(
"ANTHROPIC_API_KEY est défini : on veut l'abonnement, pas l'API. "
"Retire cette variable (utilise CLAUDE_CODE_OAUTH_TOKEN)."
)
def _build_prompt(self, snapshots: list[PairSnapshot]) -> str:
payload = {
"instruction": "Produis un biais directionnel pour chaque paire ci-dessous.",
"market_snapshot": [s.to_dict() for s in snapshots],
"pairs": [s.pair for s in snapshots],
}
return (
"Voici l'instantané de marché (JSON). Si le MCP TradingView est disponible, "
"tu peux l'utiliser pour confirmer (RSI/MACD/Bollinger).\n\n"
+ json.dumps(payload, ensure_ascii=False)
)
def get_biases(self, snapshots: list[PairSnapshot]) -> MarketBiasBatch:
if not snapshots:
return MarketBiasBatch(biases=[])
system_prompt = (
_SYSTEM_PROMPT_FILE.read_text(encoding="utf-8")
if _SYSTEM_PROMPT_FILE.exists()
else ""
)
cmd = [
"claude",
"-p",
self._build_prompt(snapshots),
"--model",
self.model,
"--output-format",
"json",
"--json-schema",
json.dumps(MarketBiasBatch.json_schema()),
"--max-turns",
str(self.max_turns),
]
if system_prompt:
cmd += ["--append-system-prompt", system_prompt]
if self.mcp_config:
cmd += ["--mcp-config", self.mcp_config, "--strict-mcp-config"]
if self.allowed_tools:
cmd += ["--allowedTools", self.allowed_tools]
else:
# Pas de MCP : on coupe les outils intégrés (analyse pure sur l'instantané).
cmd += [
"--disallowedTools",
"Bash,Read,Write,Edit,WebSearch,WebFetch,Glob,Grep,Task",
]
try:
proc = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=self.timeout_s,
env=os.environ.copy(),
)
except subprocess.TimeoutExpired as exc:
raise ClaudeError(f"claude -p a dépassé {self.timeout_s}s") from exc
if proc.returncode != 0:
raise ClaudeError(
f"claude -p code {proc.returncode}: {proc.stderr.strip()[:500]}"
)
return self._parse(proc.stdout)
@staticmethod
def _parse(stdout: str) -> MarketBiasBatch:
try:
envelope = json.loads(stdout)
except json.JSONDecodeError as exc:
raise ClaudeError(f"Sortie claude non-JSON: {stdout[:300]}") from exc
if envelope.get("is_error"):
raise ClaudeError(
f"claude erreur ({envelope.get('subtype')}): "
f"{envelope.get('errors') or envelope.get('result')}"
)
# Préférence : sortie structurée validée par le schéma.
structured = envelope.get("structured_output")
if isinstance(structured, dict):
return MarketBiasBatch.model_validate(structured)
# Repli : extraire un objet JSON du champ `result`.
result = envelope.get("result", "")
obj = _extract_json_object(result)
if obj is not None:
return MarketBiasBatch.model_validate(obj)
raise ClaudeError(f"Pas de sortie structurée exploitable: {result[:300]}")
def _extract_json_object(text: str) -> Optional[dict]:
start = text.find("{")
end = text.rfind("}")
if start == -1 or end == -1 or end <= start:
return None
try:
return json.loads(text[start : end + 1])
except json.JSONDecodeError:
return None

View File

@@ -0,0 +1,92 @@
"""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

58
ai_analyzer/models.py Normal file
View File

@@ -0,0 +1,58 @@
"""Schémas de données partagés (Pydantic) pour la couche IA."""
from __future__ import annotations
from typing import Literal, Optional
from pydantic import BaseModel, Field
Direction = Literal["bullish", "bearish", "neutral"]
class MarketBias(BaseModel):
"""Biais de marché produit par Claude pour une paire."""
pair: str = Field(..., description="Paire, ex. 'BTC/USDT'")
direction: Direction = Field(..., description="Sens du biais")
confidence: float = Field(..., ge=0.0, le=1.0, description="Confiance [0..1]")
rationale: str = Field(..., description="Justification courte et factuelle")
key_support: Optional[float] = Field(None, description="Support clé (prix)")
key_resistance: Optional[float] = Field(None, description="Résistance clé (prix)")
class MarketBiasBatch(BaseModel):
"""Lot de biais (un appel Claude couvre toutes les paires)."""
biases: list[MarketBias]
def by_pair(self) -> dict[str, MarketBias]:
return {b.pair: b for b in self.biases}
@staticmethod
def json_schema() -> dict:
"""Schéma JSON passé à `claude -p --json-schema` (sous-ensemble supporté)."""
return {
"type": "object",
"additionalProperties": False,
"required": ["biases"],
"properties": {
"biases": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": False,
"required": ["pair", "direction", "confidence", "rationale"],
"properties": {
"pair": {"type": "string"},
"direction": {
"type": "string",
"enum": ["bullish", "bearish", "neutral"],
},
"confidence": {"type": "number"},
"rationale": {"type": "string"},
"key_support": {"type": "number"},
"key_resistance": {"type": "number"},
},
},
}
},
}

View File

@@ -0,0 +1,3 @@
ccxt>=4.4
redis>=5.0
pydantic>=2.6

View File

@@ -0,0 +1,89 @@
"""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

View File

@@ -0,0 +1,13 @@
Tu es un analyste de marché crypto pour un bot de trading qui opère en simulation (dry-run).
Ton rôle : à partir d'un instantané de marché compact (et, si disponibles, des outils TradingView via MCP : RSI, MACD, Bollinger, screening), produire un **biais directionnel** par paire.
Règles :
- Sois factuel et prudent. Si le signal est ambigu, réponds `neutral` avec une confiance basse.
- `confidence` reflète la force/clarté du signal technique (0 = aucune conviction, 1 = signal très net).
- `rationale` : une phrase, concrète, citant les éléments observés (tendance, momentum, niveaux). Pas de blabla.
- Si des niveaux clairs existent, renseigne `key_support` / `key_resistance` (prix).
- N'invente pas de données. Tu n'exécutes aucun trade ; tu ne fais que qualifier le contexte.
- Couvre TOUTES les paires demandées, dans le même ordre.
Réponds UNIQUEMENT via la sortie structurée demandée (schéma JSON), sans texte additionnel.