a3fb486477
- database.py: add delete_ai_report(), delete_reasoning_state(), delete_kb_entry()
- reasoning.py: DELETE /api/reasoning/reports/{id}
- knowledge.py: DELETE /api/knowledge/history/{id} and /entries/{id}
- useApi.ts: useDeleteAiReport, useDeleteReasoningState, useDeleteKbEntry hooks
- RapportIA.tsx: trash icon on hover in archived reports sidebar
- SuperContexte.tsx: trash icon on hover for history versions and KB entries;
both propagate onDelete through CategorySection down to KbEntry
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
382 lines
15 KiB
Python
382 lines
15 KiB
Python
"""
|
|
AI Reasoning Traces — store and query the full reasoning chain behind each trade.
|
|
|
|
Endpoints:
|
|
GET /api/reasoning/postmortem/{trade_id} — reasoning chain (no GPT call)
|
|
POST /api/reasoning/postmortem/{trade_id}/analyze — GPT-4o post-mortem analysis
|
|
"""
|
|
import json
|
|
import logging
|
|
import os
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
|
|
from services.database import (
|
|
get_config,
|
|
get_ai_report,
|
|
get_mtm_trades_with_traces,
|
|
get_pattern_scoring_history,
|
|
get_scoring_trace,
|
|
get_suggestion_trace,
|
|
get_trade_entry_by_id,
|
|
list_ai_reports,
|
|
save_ai_report,
|
|
delete_ai_report,
|
|
_trade_maturity,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/reasoning", tags=["reasoning"])
|
|
|
|
|
|
# ── Helpers ───────────────────────────────────────────────────────────────────
|
|
|
|
def _bucket_summary(buckets: list) -> str:
|
|
lines = []
|
|
for b in buckets:
|
|
pct = round(b.get("score", 0) / b.get("max", 1) * 100) if b.get("max") else 0
|
|
lines.append(f" {b.get('label', b.get('id'))}: {b.get('score')}/{b.get('max')} ({pct}%) — {(b.get('comment') or '')[:90]}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _rankings_summary(rankings: list) -> str:
|
|
lines = []
|
|
for r in rankings:
|
|
delta = r.get("score_delta", 0)
|
|
sign = "+" if delta >= 0 else ""
|
|
lines.append(f" {r.get('underlying')} {r.get('strategy')} — delta {sign}{delta} | {(r.get('rationale') or '')[:80]}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ── Endpoints ─────────────────────────────────────────────────────────────────
|
|
|
|
@router.get("/postmortem/{trade_id}")
|
|
def get_postmortem(trade_id: int):
|
|
"""
|
|
Return the full AI reasoning chain for a logged trade:
|
|
- why the pattern was suggested (suggestion trace)
|
|
- why it was scored at that level (scoring trace with pillar breakdown)
|
|
- score evolution across cycles (trend)
|
|
"""
|
|
trade = get_trade_entry_by_id(trade_id)
|
|
if not trade:
|
|
raise HTTPException(404, f"Trade {trade_id} not found")
|
|
|
|
scoring_trace = get_scoring_trace(trade["run_id"], trade["pattern_id"])
|
|
suggestion_trace = get_suggestion_trace(trade["pattern_id"])
|
|
score_history = get_pattern_scoring_history(trade["pattern_id"], limit=8)
|
|
|
|
return {
|
|
"trade": trade,
|
|
"scoring_context": scoring_trace,
|
|
"suggestion_context": suggestion_trace,
|
|
"score_history": [
|
|
{
|
|
"run_id": t["run_id"],
|
|
"created_at": t["created_at"],
|
|
"score": t["output"].get("score"),
|
|
"key_catalyst": t["output"].get("key_catalyst"),
|
|
"macro_dominant": t["macro_dominant"],
|
|
"geo_score": t["geo_score"],
|
|
"summary": t["output"].get("summary"),
|
|
}
|
|
for t in score_history
|
|
],
|
|
}
|
|
|
|
|
|
@router.post("/postmortem/{trade_id}/analyze")
|
|
def analyze_postmortem(trade_id: int):
|
|
"""
|
|
Ask GPT-4o to explain why a trade did/didn't work based on the full reasoning chain.
|
|
Returns a structured analysis with diagnostic, lessons, and next-cycle recommendations.
|
|
"""
|
|
ai_key = get_config("openai_api_key") or ""
|
|
if not ai_key:
|
|
raise HTTPException(400, "Clé OpenAI non configurée")
|
|
os.environ["OPENAI_API_KEY"] = ai_key
|
|
|
|
from services.ai_analyzer import _chat
|
|
|
|
trade = get_trade_entry_by_id(trade_id)
|
|
if not trade:
|
|
raise HTTPException(404, f"Trade {trade_id} not found")
|
|
|
|
scoring_trace = get_scoring_trace(trade["run_id"], trade["pattern_id"])
|
|
suggestion_trace = get_suggestion_trace(trade["pattern_id"])
|
|
score_history = get_pattern_scoring_history(trade["pattern_id"], limit=5)
|
|
|
|
scoring_out = scoring_trace["output"] if scoring_trace else {}
|
|
scoring_ctx = scoring_trace["input_context"] if scoring_trace else {}
|
|
suggestion_out = suggestion_trace["output"] if suggestion_trace else {}
|
|
|
|
buckets_text = _bucket_summary(scoring_out.get("buckets", []))
|
|
rankings_text = _rankings_summary(scoring_out.get("trade_rankings", []))
|
|
|
|
score_trend = " → ".join(
|
|
f"{t['output'].get('score', '?')}/100 ({t['macro_dominant'] or '?'} régime, géo {t['geo_score'] or '?'})"
|
|
for t in reversed(score_history)
|
|
)
|
|
|
|
prompt = f"""Tu es un stratège macro-géopolitique senior qui analyse le post-mortem d'un trade options.
|
|
|
|
═══ TRADE ANALYSÉ ═══
|
|
Pattern : {trade.get("pattern_name")}
|
|
Instrument : {trade.get("underlying")} — {trade.get("strategy")}
|
|
Entrée : {trade.get("entry_date")} @ {trade.get("entry_price") or "N/A"}
|
|
Score entrée: {trade.get("score_at_entry")}/100 | Trade Score: {trade.get("trade_score") or "N/A"} | EV nette: {trade.get("ev_net") or "N/A"}
|
|
Profil : {trade.get("matched_profile")} | Gain prévu: {trade.get("expected_move_pct") or "N/A"}%
|
|
|
|
═══ CONTEXTE AU MOMENT DU SCORING ═══
|
|
Régime macro : {scoring_trace.get("macro_dominant") if scoring_trace else "N/A"} | Biais asset: {scoring_ctx.get("asset_bias", "N/A")}
|
|
Scores macro : {json.dumps(scoring_ctx.get("macro_scores", {}), ensure_ascii=False)}
|
|
Risque géo : {scoring_trace.get("geo_score") if scoring_trace else "N/A"}/100
|
|
Gain prévu : {scoring_ctx.get("expected_move_pct") or "N/A"}%
|
|
|
|
═══ POURQUOI CE PATTERN A ÉTÉ CRÉÉ ═══
|
|
{suggestion_out.get("macro_fit") or "N/A"}
|
|
{suggestion_out.get("description") or ""}
|
|
|
|
═══ SCORE DÉTAILLÉ PAR PILIER ═══
|
|
Score global : {scoring_out.get("score", 0)}/100 (confiance {scoring_out.get("confidence", 0)}%)
|
|
{buckets_text or "Non disponible"}
|
|
Catalyseur clé : {scoring_out.get("key_catalyst") or "N/A"}
|
|
Synthèse : {scoring_out.get("summary") or "N/A"}
|
|
Contra-signal fort : {"OUI" if scoring_out.get("has_strong_contra") else "non"}
|
|
|
|
═══ CLASSEMENT DES TRADES AU SCORING ═══
|
|
{rankings_text or "Non disponible"}
|
|
|
|
═══ ÉVOLUTION DU SCORE DANS LE TEMPS ═══
|
|
{score_trend or "Premier scoring — pas d'historique"}
|
|
|
|
Analyse ce trade en JSON :
|
|
{{
|
|
"diagnostic": "<2-3 phrases: qu'explique la performance (bonne ou mauvaise) de ce trade ?>",
|
|
"what_worked": "<ce qui était correct dans l'analyse initiale>",
|
|
"what_missed": "<ce que l'IA a sous/sur-estimé, ou n'a pas anticipé>",
|
|
"regime_alignment": "<le régime macro était-il vraiment favorable ? a-t-il évolué depuis ?>",
|
|
"contra_assessment": "<les contra-signals détectés étaient-ils le vrai risque ? ou un faux signal ?>",
|
|
"lesson": "<1 règle précise à retenir pour scorer ce type de pattern plus finement>",
|
|
"next_cycle": "<comment enrichir le contexte et les critères pour ce pattern dans les prochains cycles ?>"
|
|
}}"""
|
|
|
|
try:
|
|
result = _chat(
|
|
"Tu es un stratège macro-géopolitique senior. Post-mortem concis et actionnable. JSON uniquement.",
|
|
prompt,
|
|
model="gpt-4o",
|
|
json_mode=True,
|
|
max_tokens=900,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"[Postmortem] GPT-4o call failed: {e}")
|
|
raise HTTPException(503, "GPT-4o indisponible")
|
|
|
|
if not result:
|
|
raise HTTPException(503, "GPT-4o n'a pas retourné de réponse")
|
|
|
|
return {
|
|
"trade_id": trade_id,
|
|
"trade": trade,
|
|
"scoring_context": scoring_trace,
|
|
"suggestion_context": suggestion_trace,
|
|
"analysis": result,
|
|
}
|
|
|
|
|
|
# ── Portfolio AI Report ────────────────────────────────────────────────────────
|
|
|
|
def _trade_summary_block(label: str, trades: list) -> str:
|
|
if not trades:
|
|
return f"{label} : aucun trade pricé"
|
|
lines = [f"{label} :"]
|
|
for t in trades:
|
|
pnl = t.get("pnl_pct")
|
|
sc = t.get("scoring_context") or {}
|
|
sc_out = sc.get("output", {}) if isinstance(sc, dict) else {}
|
|
sg = t.get("suggestion_context") or {}
|
|
sg_out = sg.get("output", {}) if isinstance(sg, dict) else {}
|
|
macro = sc.get("macro_dominant") if isinstance(sc, dict) else "?"
|
|
geo = sc.get("geo_score") if isinstance(sc, dict) else "?"
|
|
catalyst = sc_out.get("key_catalyst") or "N/A"
|
|
macro_fit = sg_out.get("macro_fit") or sg_out.get("description") or "N/A"
|
|
trend = " → ".join(str(s) for s in (t.get("score_trend") or [])) or "N/A"
|
|
buckets = sc_out.get("buckets", [])
|
|
weak = [b.get("label", b.get("id", "")) for b in buckets if b.get("max") and b.get("score", 0) / b["max"] < 0.4]
|
|
lines.append(
|
|
f" • {t.get('pattern_name')} | {t.get('underlying')} {t.get('strategy')}"
|
|
f" | P&L {'+' if (pnl or 0) >= 0 else ''}{(pnl or 0):.1f}%"
|
|
f" | Score entrée {t.get('score_at_entry')}/100 | Régime {macro} | Géo {geo}"
|
|
f"\n Thèse : {macro_fit[:120]}"
|
|
f"\n Catalyseur : {catalyst}"
|
|
f"\n Trend score : {trend}"
|
|
+ (f"\n Piliers faibles : {', '.join(weak)}" if weak else "")
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
|
|
@router.get("/portfolio-report")
|
|
def get_portfolio_report_data(days: int = 90):
|
|
"""Return raw MTM + traces data (no GPT-4o call) for the report page."""
|
|
data = get_mtm_trades_with_traces(days=days, limit_movers=5)
|
|
return data
|
|
|
|
|
|
@router.post("/portfolio-report/generate")
|
|
def generate_portfolio_report(days: int = 90):
|
|
"""
|
|
Generate a GPT-4o AI report: key highlights, explanations for top movers,
|
|
macro regime assessment, and actionable next-cycle recommendations.
|
|
"""
|
|
ai_key = get_config("openai_api_key") or ""
|
|
if not ai_key:
|
|
raise HTTPException(400, "Clé OpenAI non configurée")
|
|
os.environ["OPENAI_API_KEY"] = ai_key
|
|
|
|
from services.ai_analyzer import _chat
|
|
|
|
from datetime import date as _date
|
|
|
|
data = get_mtm_trades_with_traces(days=days, limit_movers=10)
|
|
all_trades = data.get("all_trades", [])
|
|
|
|
# Classify every trade by maturity
|
|
def _enrich_maturity(t: dict) -> dict:
|
|
try:
|
|
dh = (_date.today() - _date.fromisoformat(t["entry_date"])).days
|
|
except Exception:
|
|
dh = 0
|
|
return {**t, "days_held": dh, "maturity": _trade_maturity(dh, t.get("horizon_days") or 90)}
|
|
|
|
enriched = [_enrich_maturity(t) for t in all_trades if t.get("pnl_pct") is not None]
|
|
trop_tot = [t for t in enriched if t["maturity"]["status"] == "trop_tot"]
|
|
en_cours = [t for t in enriched if t["maturity"]["status"] == "debut"]
|
|
matures = [t for t in enriched if t["maturity"]["status"] in ("mature", "fin_horizon")]
|
|
|
|
# Sort by P&L for report
|
|
winners = sorted(matures, key=lambda t: t.get("pnl_pct", 0), reverse=True)[:5]
|
|
losers = sorted(matures, key=lambda t: t.get("pnl_pct", 0))[:5]
|
|
avg_pnl = (sum(t.get("pnl_pct", 0) for t in matures) / len(matures)) if matures else None
|
|
avg_str = f"{avg_pnl:+.1f}%" if avg_pnl is not None else "N/A"
|
|
|
|
def _tline(t: dict) -> str:
|
|
mat = t["maturity"]
|
|
return (f" {t.get('underlying','?')} {t.get('strategy','?')} "
|
|
f"P&L={t.get('pnl_pct',0):+.1f}% [{mat['readable']}] "
|
|
f"score={t.get('score_at_entry','?')}")
|
|
|
|
mature_wins = "\n".join(_tline(t) for t in winners) or " Aucun"
|
|
mature_loss = "\n".join(_tline(t) for t in losers) or " Aucun"
|
|
early_lines = "\n".join(_tline(t) for t in (trop_tot + en_cours)[:8]) or " Aucun"
|
|
|
|
prompt = f"""Tu es un stratège macro-géopolitique senior. Génère un rapport synthétique sur notre portefeuille options.
|
|
|
|
⚠️ RÈGLE FONDAMENTALE DE TIMING :
|
|
Nos trades sont des options de 30 à 90 jours. Un trade vieux de 3 jours n'apporte AUCUNE information sur sa performance finale.
|
|
Tu dois tirer des leçons UNIQUEMENT des trades MATURES (≥35% de l'horizon écoulé).
|
|
Les trades IMMATURES (< 35%) sont listés pour transparence — n'en tire aucune conclusion de performance.
|
|
|
|
═══ STATISTIQUES GLOBALES ═══
|
|
Période : {days}j | Total trades: {len(all_trades)} | Pricés: {data['priced_count']}
|
|
MATURES (signal fiable): {len(matures)} — P&L moyen: {avg_str}
|
|
IMMATURES (trop tôt): {len(trop_tot) + len(en_cours)}
|
|
|
|
═══ MATURES — TOP GAINS (signal fiable — tire des leçons ici) ═══
|
|
{mature_wins}
|
|
|
|
═══ MATURES — TOP PERTES (signal fiable — tire des leçons ici) ═══
|
|
{mature_loss}
|
|
|
|
═══ EN COURS / IMMATURES (ne pas juger la performance) ═══
|
|
{early_lines}
|
|
|
|
Génère un rapport JSON structuré basé UNIQUEMENT sur les trades matures :
|
|
{{
|
|
"headline": "<1 phrase résumant la performance des trades matures>",
|
|
"regime_assessment": "<le régime macro a-t-il bien servi nos thèses matures ?>",
|
|
"winners_analysis": "<pourquoi les trades matures gagnants ont marché — 3-4 phrases>",
|
|
"losers_analysis": "<pourquoi les trades matures perdants ont déçu — 3-4 phrases>",
|
|
"key_lessons": ["<leçon 1 des matures>", "<leçon 2>", "<leçon 3>"],
|
|
"blind_spots": "<ce que le scoring n'a pas bien capturé sur les trades matures>",
|
|
"next_cycle_priorities": "<3 priorités concrètes pour les prochains cycles>",
|
|
"risk_watch": "<1-2 risques à surveiller>",
|
|
"timing_note": "<observation sur les trades immatures — signaux à surveiller sans jugement>"
|
|
}}"""
|
|
|
|
try:
|
|
result = _chat(
|
|
"Tu es un stratège macro senior. Rapport synthétique et actionnable. JSON uniquement.",
|
|
prompt,
|
|
model="gpt-4o",
|
|
json_mode=True,
|
|
max_tokens=1400,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"[PortfolioReport] GPT-4o call failed: {e}")
|
|
raise HTTPException(503, "GPT-4o indisponible")
|
|
|
|
if not result:
|
|
raise HTTPException(503, "GPT-4o n'a pas retourné de réponse")
|
|
|
|
stats = {
|
|
"total_trades": len(all_trades),
|
|
"priced_count": data["priced_count"],
|
|
"avg_pnl_pct": avg_pnl,
|
|
"mature_count": len(matures),
|
|
"early_count": len(trop_tot) + len(en_cours),
|
|
}
|
|
|
|
report_id = save_ai_report(
|
|
days=days,
|
|
stats=stats,
|
|
winners=winners,
|
|
losers=losers,
|
|
report=result,
|
|
)
|
|
|
|
return {
|
|
"id": report_id,
|
|
"days": days,
|
|
"stats": stats,
|
|
"winners": winners,
|
|
"losers": losers,
|
|
"report": result,
|
|
}
|
|
|
|
|
|
@router.get("/reports")
|
|
def list_reports(report_type: str = "portfolio", limit: int = 20):
|
|
"""List archived AI reports (newest first), summary only."""
|
|
reports = list_ai_reports(report_type=report_type, limit=limit)
|
|
return {
|
|
"reports": [
|
|
{
|
|
"id": r["id"],
|
|
"days": r["days"],
|
|
"created_at": r["created_at"],
|
|
"stats": r["stats"],
|
|
"headline": r["report"].get("headline", ""),
|
|
}
|
|
for r in reports
|
|
]
|
|
}
|
|
|
|
|
|
@router.get("/reports/{report_id}")
|
|
def get_report(report_id: int):
|
|
"""Retrieve a full archived AI report by ID."""
|
|
report = get_ai_report(report_id)
|
|
if not report:
|
|
raise HTTPException(404, f"Report {report_id} not found")
|
|
return report
|
|
|
|
|
|
@router.delete("/reports/{report_id}")
|
|
def delete_report(report_id: int):
|
|
"""Delete an archived AI report by ID."""
|
|
deleted = delete_ai_report(report_id)
|
|
if not deleted:
|
|
raise HTTPException(404, f"Report {report_id} not found")
|
|
return {"deleted": True, "id": report_id}
|