2308 lines
113 KiB
Python
2308 lines
113 KiB
Python
"""
|
||
Auto-cycle orchestration — runs every N hours (configurable).
|
||
|
||
Cycle steps:
|
||
1. Fetch current context (news, quotes, macro, geo)
|
||
2. Ask GPT-4o to suggest new patterns
|
||
3. Filter: keep only those with Jaccard keyword similarity < threshold vs existing
|
||
4. Save filtered patterns to DB
|
||
5. Score ALL patterns (existing + new)
|
||
6. Log: pattern scores, trade entry prices, geo alert, macro snapshot
|
||
7. Generate GPT-4o commentary: why are top/bottom trades performing this way?
|
||
8. Update cycle_runs with results + commentary
|
||
"""
|
||
import logging
|
||
import threading
|
||
import uuid
|
||
from datetime import datetime, timedelta
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── Global scheduler state ────────────────────────────────────────────────────
|
||
|
||
_stop_event = threading.Event()
|
||
_cycle_thread: Optional[threading.Thread] = None
|
||
_cycle_lock = threading.Lock() # prevents concurrent cycles
|
||
_current_status: Dict[str, Any] = {
|
||
"running": False,
|
||
"last_run_id": None,
|
||
"last_run_at": None,
|
||
"next_run_at": None,
|
||
"enabled": False,
|
||
"interval_hours": 3,
|
||
}
|
||
|
||
|
||
# ── Helpers ───────────────────────────────────────────────────────────────────
|
||
|
||
def _jaccard(a: List[str], b: List[str]) -> float:
|
||
sa = {x.lower() for x in (a or [])}
|
||
sb = {x.lower() for x in (b or [])}
|
||
if not sa and not sb:
|
||
return 0.0
|
||
union = sa | sb
|
||
return len(sa & sb) / len(union) if union else 0.0
|
||
|
||
|
||
def _max_similarity_vs_existing(candidate_kws: List[str], existing: List[Dict]) -> float:
|
||
return max((_jaccard(candidate_kws, p.get("keywords") or []) for p in existing), default=0.0)
|
||
|
||
|
||
_EMBED_SIM_THRESHOLD = 0.75 # cosine threshold to consider two patterns as duplicates — fallback default, see CYCLE_STEP_CATALOG["duplicate_detection"]
|
||
|
||
|
||
def _is_duplicate_pattern(
|
||
candidate: Dict,
|
||
existing: List[Dict],
|
||
api_key: str,
|
||
jaccard_threshold: float = 0.30,
|
||
embed_threshold: float = _EMBED_SIM_THRESHOLD,
|
||
) -> bool:
|
||
"""
|
||
Returns True if the candidate is too similar to an existing pattern.
|
||
Tries cosine embeddings (Sprint 4.3) with Jaccard fallback.
|
||
"""
|
||
if not existing:
|
||
return False
|
||
|
||
# Tentative embedding cosinus
|
||
if api_key:
|
||
text = ((candidate.get("name") or "") + " " + (candidate.get("description") or "")).strip()
|
||
if text:
|
||
try:
|
||
from services.database import max_cosine_similarity_vs_existing
|
||
sim = max_cosine_similarity_vs_existing(
|
||
candidate_text=text,
|
||
existing_patterns=existing,
|
||
api_key=api_key,
|
||
candidate_id=candidate.get("id"),
|
||
)
|
||
if sim > 0: # embedding worked
|
||
return sim >= embed_threshold
|
||
except Exception as _emb_err:
|
||
logger.debug(f"[Cycle] Embedding failed, fallback Jaccard: {_emb_err}")
|
||
|
||
# Fallback Jaccard
|
||
sim = _max_similarity_vs_existing(candidate.get("keywords") or [], existing)
|
||
return sim >= jaccard_threshold
|
||
|
||
|
||
# ── Cycle step configuration — self-describing catalog + DB-backed overrides ──
|
||
# Mirrors the SIGNAL_CATALOG pattern already used for AI Desks (routers/ai_desks.py):
|
||
# one declarative entry per knob so Config.tsx can render a generic form instead
|
||
# of hand-coded sliders. Stored as one JSON blob under config key
|
||
# "cycle_step_config" (same convention as exit_defaults/analysis_config), merged
|
||
# with these defaults at read time — adding a new knob here never requires a
|
||
# DB migration.
|
||
|
||
CYCLE_STEP_CATALOG: List[Dict[str, Any]] = [
|
||
{"id": "portfolio_monitor", "label": "Portfolio Monitor", "group": "portfolio",
|
||
"description": "Alerte IA quand le risque du portefeuille simulé dépasse un seuil (conflits, concentration).",
|
||
"params": {"enabled": {"type": "bool", "default": True}}},
|
||
{"id": "watchlist_auto_add", "label": "Ajout auto à la watchlist", "group": "portfolio",
|
||
"description": "Ajoute automatiquement les nouveaux tickers détectés par le cycle à la watchlist.",
|
||
"params": {"enabled": {"type": "bool", "default": True}}},
|
||
{"id": "regime_clustering", "label": "Clustering de régime macro", "group": "ai",
|
||
"description": "Reclassifie le régime macro courant par clustering.",
|
||
"params": {
|
||
"enabled": {"type": "bool", "default": True},
|
||
"n_clusters": {"type": "int", "label": "Nb clusters", "default": 4, "min": 2, "max": 8},
|
||
"lookback_days": {"type": "int", "label": "Lookback (j)", "default": 180, "min": 30, "max": 365},
|
||
}},
|
||
{"id": "knowledge_synthesis", "label": "Synthèse de connaissances", "group": "ai",
|
||
"description": "Résume les derniers rapports IA en enseignements exploitables.",
|
||
"params": {
|
||
"enabled": {"type": "bool", "default": True},
|
||
"min_hours_between": {"type": "int", "label": "Délai min entre synthèses (h)", "default": 6, "min": 1, "max": 48},
|
||
}},
|
||
{"id": "institutional_absorption", "label": "Absorption rapports institutionnels", "group": "data",
|
||
"description": "Fenêtre de suivi de l'absorption marché d'un rapport institutionnel.",
|
||
"params": {"lookback_days": {"type": "int", "label": "Lookback (j)", "default": 14, "min": 3, "max": 60}}},
|
||
{"id": "var_snapshot", "label": "Snapshot VaR", "group": "portfolio",
|
||
"description": "Paramètres du calcul de Value-at-Risk dans le rapport de cycle.",
|
||
"params": {
|
||
"confidence": {"type": "float", "label": "Confiance", "default": 0.95, "min": 0.80, "max": 0.99},
|
||
"horizon_days": {"type": "int", "label": "Horizon (j)", "default": 1, "min": 1, "max": 30},
|
||
"lookback_days": {"type": "int", "label": "Lookback (j)", "default": 252, "min": 60, "max": 504},
|
||
"default_iv": {"type": "float", "label": "IV par défaut", "default": 0.20, "min": 0.05, "max": 1.0},
|
||
}},
|
||
{"id": "duplicate_detection", "label": "Détection de doublons de patterns", "group": "ai",
|
||
"description": "Seuil de similarité cosinus (embeddings) pour éviter de recréer un pattern existant.",
|
||
"params": {"embedding_threshold": {"type": "float", "label": "Seuil embedding", "default": 0.75, "min": 0.5, "max": 0.95}}},
|
||
{"id": "price_snapshot_retention", "label": "Rétention snapshots de prix", "group": "data",
|
||
"description": "Durée de conservation des snapshots de prix intra-cycle.",
|
||
"params": {"days": {"type": "int", "label": "Jours", "default": 14, "min": 1, "max": 90}}},
|
||
{"id": "iv_context", "label": "Contexte IV / mouvements récents", "group": "data",
|
||
"description": "Fenêtre de trades utilisée pour le contexte IV (utilisée à 3 endroits du cycle).",
|
||
"params": {"lookback_days": {"type": "int", "label": "Jours", "default": 90, "min": 7, "max": 365}}},
|
||
{"id": "absorption_detection", "label": "Détection d'absorption de prix", "group": "data",
|
||
"description": "Fenêtre d'âge pour la détection d'absorption de prix (Phase 4B).",
|
||
"params": {
|
||
"min_age_minutes": {"type": "float", "label": "Âge min (min)", "default": 30.0, "min": 5, "max": 180},
|
||
"max_age_days": {"type": "int", "label": "Âge max (j)", "default": 7, "min": 1, "max": 30},
|
||
}},
|
||
{"id": "economic_surprises", "label": "Surprises économiques (FRED)", "group": "data",
|
||
"description": "Fenêtre de lookback pour les surprises économiques récentes.",
|
||
"params": {"lookback_days": {"type": "int", "label": "Jours", "default": 60, "min": 7, "max": 180}}},
|
||
{"id": "institutional_block", "label": "Bloc institutionnel (prompt)", "group": "data",
|
||
"description": "Fenêtre des rapports institutionnels injectés dans le prompt de suggestion.",
|
||
"params": {"lookback_days": {"type": "int", "label": "Jours", "default": 7, "min": 1, "max": 30}}},
|
||
{"id": "cycle_commentary", "label": "Commentaire de fin de cycle", "group": "ai",
|
||
"description": "Fenêtre de trades récents utilisée pour le commentaire GPT-4o de fin de cycle.",
|
||
"params": {"lookback_days": {"type": "int", "label": "Jours", "default": 7, "min": 1, "max": 30}}},
|
||
]
|
||
|
||
|
||
def _default_cycle_step_config() -> Dict[str, Dict[str, Any]]:
|
||
return {step["id"]: {k: p["default"] for k, p in step["params"].items()} for step in CYCLE_STEP_CATALOG}
|
||
|
||
|
||
def get_cycle_step_config() -> Dict[str, Dict[str, Any]]:
|
||
"""Merge saved overrides (config key "cycle_step_config") over the catalog
|
||
defaults — missing keys/steps fall back to defaults, so adding a new knob
|
||
to CYCLE_STEP_CATALOG is never a breaking change for existing saved config."""
|
||
import json
|
||
from services.database import get_config
|
||
defaults = _default_cycle_step_config()
|
||
try:
|
||
saved = json.loads(get_config("cycle_step_config") or "{}")
|
||
except Exception:
|
||
saved = {}
|
||
return {step_id: {**step_defaults, **(saved.get(step_id) or {})} for step_id, step_defaults in defaults.items()}
|
||
|
||
|
||
# ── Portfolio monitor agent ───────────────────────────────────────────────────
|
||
|
||
def _run_portfolio_monitor(risk: dict, cycle_id: str) -> dict:
|
||
"""
|
||
AI agent that analyses the open simulation portfolio and generates recommendations.
|
||
Runs only when there are alerts (conflicts, concentration). Results saved to system_logs.
|
||
"""
|
||
from services.portfolio_risk import build_monitor_context
|
||
from services.database import log_system_event
|
||
from services.ai_analyzer import _chat
|
||
import json as _json
|
||
|
||
context = build_monitor_context(risk)
|
||
n_conflicts = len(risk.get("conflicts", []))
|
||
n_alerts = len(risk.get("alerts", []))
|
||
|
||
system_prompt = (
|
||
"You are a risk manager specializing in geopolitical options portfolios. "
|
||
"Analyze the state of the simulated portfolio below and generate concrete recommendations. "
|
||
"Respond in JSON with this exact schema: "
|
||
'{"assessment": "<max 2 sentences on the overall state>", '
|
||
'"actions": [{"priority": "high|medium", "type": "close_trade|rebalance|monitor", '
|
||
'"trade_id": <int or null>, "underlying": "<ticker>", "reason": "<short reason>"}], '
|
||
'"rebalance_suggestion": "<concrete rebalancing suggestion in 1 sentence>"}'
|
||
)
|
||
user_prompt = context
|
||
|
||
try:
|
||
result = _chat(system_prompt, user_prompt, model="gpt-4o-mini", json_mode=True, max_tokens=600)
|
||
if isinstance(result, dict) and "assessment" in result:
|
||
details_str = _json.dumps(result, ensure_ascii=False)
|
||
msg = f"[PortfolioMonitor] {n_conflicts} conflits, {n_alerts} alertes — {result.get('assessment', '')[:120]}"
|
||
log_system_event("WARN" if n_conflicts else "INFO", "portfolio_monitor",
|
||
msg, cycle_id=cycle_id, details=details_str)
|
||
logger.info(f"[PortfolioMonitor] AI assessment saved — {len(result.get('actions', []))} actions")
|
||
return result
|
||
except Exception as e:
|
||
logger.warning(f"[PortfolioMonitor] AI call failed: {e}")
|
||
|
||
# Fallback: log raw alerts without AI
|
||
fallback = {
|
||
"assessment": f"{n_conflicts} conflits directionnels et {n_alerts} alertes détectées dans le portefeuille simulé.",
|
||
"actions": [
|
||
{"priority": "high" if a["level"] == "danger" else "medium",
|
||
"type": "close_trade" if a["type"] == "conflict" else "monitor",
|
||
"trade_id": None, "underlying": a.get("underlying", ""),
|
||
"reason": a["message"]}
|
||
for a in risk.get("alerts", [])[:5]
|
||
],
|
||
"rebalance_suggestion": "Revérifier manuellement les positions contradictoires.",
|
||
}
|
||
log_system_event("WARN", "portfolio_monitor",
|
||
f"[PortfolioMonitor] {n_alerts} alertes (analyse IA indisponible)",
|
||
cycle_id=cycle_id, details=_json.dumps(fallback, ensure_ascii=False))
|
||
return fallback
|
||
|
||
|
||
# ── Core cycle logic ──────────────────────────────────────────────────────────
|
||
|
||
def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
|
||
"""
|
||
Execute one full auto-cycle. Thread-safe (skips if already running).
|
||
Returns a summary dict.
|
||
"""
|
||
if not _cycle_lock.acquire(blocking=False):
|
||
logger.info("Auto-cycle skipped — another cycle is already running")
|
||
return {"skipped": True, "reason": "already_running"}
|
||
|
||
run_id = datetime.utcnow().isoformat()
|
||
summary: Dict[str, Any] = {
|
||
"run_id": run_id,
|
||
"trigger": trigger,
|
||
"patterns_suggested": 0,
|
||
"patterns_added": 0,
|
||
"patterns_scored": 0,
|
||
"geo_score": None,
|
||
"dominant_regime": None,
|
||
"commentary": None,
|
||
"status": "error",
|
||
}
|
||
|
||
try:
|
||
from services.database import (
|
||
get_config, get_custom_patterns, save_custom_pattern,
|
||
save_pattern_scores, log_macro_regime, log_geo_alert, log_trade_entries,
|
||
add_cycle_run, update_cycle_run, save_reasoning_trace,
|
||
get_latest_portfolio_lessons, log_system_event,
|
||
get_last_completed_cycle_ts, save_cycle_context_snapshot,
|
||
purge_old_price_snapshots,
|
||
)
|
||
from services.data_fetcher import fetch_geo_news, get_all_quotes, get_macro_gauges, score_macro_scenarios
|
||
from services.geo_analyzer import compute_geo_risk_score
|
||
from services.ai_analyzer import (
|
||
suggest_patterns_from_market_context, score_patterns_with_context,
|
||
ai_score_news_batch, ai_score_geo_risk, _chat, DEFAULT_ANALYSIS_TEMPLATE,
|
||
)
|
||
from services.portfolio_context import (
|
||
get_open_trades_with_moves, get_portfolio_concentration, build_portfolio_context_block,
|
||
)
|
||
|
||
# KB confidence decay (non-blocking)
|
||
try:
|
||
from services.database import decay_kb_confidence
|
||
_decayed = decay_kb_confidence()
|
||
if _decayed:
|
||
logger.info(f"[Cycle {run_id[:16]}] KB decay: {_decayed} entrée(s) mise(s) à jour")
|
||
except Exception as _e:
|
||
logger.warning(f"[Cycle] KB decay failed (non-blocking): {_e}")
|
||
|
||
# Check AI key
|
||
ai_key = get_config("openai_api_key") or ""
|
||
if not ai_key:
|
||
logger.warning("Auto-cycle: no OpenAI key configured, skipping AI steps")
|
||
return {**summary, "status": "no_ai_key"}
|
||
|
||
import os
|
||
os.environ["OPENAI_API_KEY"] = ai_key
|
||
|
||
sim_threshold = float(get_config("auto_cycle_similarity_threshold") or "0.30")
|
||
step_cfg = get_cycle_step_config()
|
||
|
||
add_cycle_run(run_id, trigger=trigger)
|
||
_current_status["running"] = True
|
||
_current_status["last_run_id"] = run_id
|
||
|
||
# ── Cycle meta — timing context ───────────────────────────────────────
|
||
_now = datetime.utcnow()
|
||
_last_cycle_ts_str = get_last_completed_cycle_ts()
|
||
_delta_minutes: float = 180.0 # default 3h if no prior cycle
|
||
if _last_cycle_ts_str:
|
||
try:
|
||
_last_dt = datetime.fromisoformat(_last_cycle_ts_str.replace("Z", ""))
|
||
_delta_minutes = max(1.0, (_now - _last_dt).total_seconds() / 60)
|
||
except Exception:
|
||
pass
|
||
|
||
_interval_hours = float(get_config("auto_cycle_hours") or "3")
|
||
if _delta_minutes < 90:
|
||
_calib_label = "Court terme (<2h) — signaux très récents"
|
||
elif _delta_minutes < 720:
|
||
_calib_label = f"Moyen terme ({_delta_minutes/60:.0f}h) — signaux potentiellement pas encore pricés"
|
||
else:
|
||
_calib_label = f"Long terme ({_delta_minutes/60:.0f}h) — marchés ont eu le temps d'intégrer"
|
||
|
||
# ── Market calendar context ────────────────────────────────────────
|
||
from datetime import timezone as _tz
|
||
_now_utc = _now.replace(tzinfo=_tz.utc) if _now.tzinfo is None else _now.astimezone(_tz.utc)
|
||
_weekday = _now_utc.weekday() # 0=Mon … 6=Sun
|
||
_day_names = ["Lundi", "Mardi", "Mercredi", "Jeudi", "Vendredi", "Samedi", "Dimanche"]
|
||
_is_weekend = _weekday >= 5
|
||
# CME equity/energy/metals futures: closed Fri 17:00 ET → Sun 18:00 ET
|
||
# Rough ET offset (UTC-5 winter / UTC-4 summer) — good enough for intent
|
||
_et_hour = (_now_utc.hour - 5) % 24
|
||
if _is_weekend:
|
||
_market_session = "closed"
|
||
_market_note = f"{_day_names[_weekday]} — marchés US et CME fermés. Derniers prix disponibles: vendredi clôture."
|
||
elif _et_hour < 4:
|
||
_market_session = "overnight"
|
||
_market_note = "Nuit US (overnight) — liquidité réduite, spreads larges."
|
||
elif _et_hour < 9:
|
||
_market_session = "pre_market"
|
||
_market_note = "Pré-marché US — prix indicatifs, faible liquidité."
|
||
elif _et_hour < 16:
|
||
_market_session = "open"
|
||
_market_note = "Marché US ouvert."
|
||
else:
|
||
_market_session = "after_hours"
|
||
_market_note = "After-hours US — prix indicatifs, liquidité réduite."
|
||
|
||
_budget_eur = float(get_config("trade_budget_eur") or "5000")
|
||
_horizon_min = int(get_config("preferred_horizon_min") or "30")
|
||
_horizon_max = int(get_config("preferred_horizon_max") or "180")
|
||
|
||
cycle_meta = {
|
||
"current_cycle_ts": _now.isoformat(),
|
||
"last_cycle_ts": _last_cycle_ts_str,
|
||
"delta_minutes": round(_delta_minutes, 1),
|
||
"interval_hours": _interval_hours,
|
||
"calibration_label": _calib_label,
|
||
"day_of_week": _day_names[_weekday],
|
||
"is_weekend": _is_weekend,
|
||
"market_session": _market_session,
|
||
"market_note": _market_note,
|
||
"trade_budget_eur": _budget_eur,
|
||
"preferred_horizon_min": _horizon_min,
|
||
"preferred_horizon_max": _horizon_max,
|
||
}
|
||
logger.info(
|
||
f"[Cycle {run_id[:16]}] Cycle meta: delta={_delta_minutes:.0f}min depuis dernier cycle"
|
||
f" ({_last_cycle_ts_str[:16] if _last_cycle_ts_str else 'premier cycle'})"
|
||
)
|
||
|
||
# ── Step 0: Load portfolio lessons + Super Contexte ──────────────────
|
||
portfolio_lessons = get_latest_portfolio_lessons()
|
||
|
||
# Load Super Contexte (accumulated knowledge base)
|
||
try:
|
||
from services.database import get_latest_reasoning_state
|
||
_reasoning_state = get_latest_reasoning_state()
|
||
if _reasoning_state:
|
||
if portfolio_lessons is None:
|
||
portfolio_lessons = {}
|
||
portfolio_lessons["super_context"] = (
|
||
f"[Super Contexte v{_reasoning_state.get('version',1)} "
|
||
f"du {(_reasoning_state.get('created_at','')[:16])}]\n"
|
||
+ _reasoning_state.get("narrative", "")[:800]
|
||
)
|
||
_synth = _reasoning_state.get("synthesis") or {}
|
||
portfolio_lessons["strategic_priorities"] = _synth.get("strategic_priorities", [])
|
||
portfolio_lessons["recurring_mistakes"] = [
|
||
m.get("mistake", "") for m in _synth.get("recurring_mistakes", [])[:3]
|
||
]
|
||
logger.info(
|
||
f"[Cycle {run_id[:16]}] Super Contexte v{_reasoning_state.get('version')} loaded "
|
||
f"({_reasoning_state.get('reports_used',0)} rapports, "
|
||
f"{_reasoning_state.get('trades_analyzed',0)} trades)"
|
||
)
|
||
except Exception as _e:
|
||
logger.warning(f"[Cycle] Could not load Super Contexte: {_e}")
|
||
|
||
if portfolio_lessons:
|
||
age_hours = 0
|
||
try:
|
||
from datetime import datetime as _dt
|
||
created = _dt.fromisoformat(portfolio_lessons["created_at"])
|
||
age_hours = (_dt.utcnow() - created).total_seconds() / 3600
|
||
except Exception:
|
||
pass
|
||
logger.info(
|
||
f"[Cycle {run_id[:16]}] Portfolio lessons loaded "
|
||
f"(report from {portfolio_lessons.get('created_at','?')[:10]}, "
|
||
f"{age_hours:.0f}h ago, avg_pnl={portfolio_lessons.get('stats', {}).get('avg_pnl_pct')}%)"
|
||
)
|
||
else:
|
||
logger.info(f"[Cycle {run_id[:16]}] No portfolio report yet — cycle runs without performance feedback")
|
||
|
||
# ── Step 1: Fetch context ─────────────────────────────────────────────
|
||
logger.info(f"[Cycle {run_id[:16]}] Step 1: fetching context")
|
||
from routers.geopolitical import _news_cache # type: ignore
|
||
news = _news_cache.get("data") or fetch_geo_news()
|
||
news = ai_score_news_batch(news)
|
||
_news_cache["data"] = news
|
||
|
||
algo_geo_score_obj = compute_geo_risk_score(news)
|
||
try:
|
||
geo_score_obj = ai_score_geo_risk(news, algo_geo_score_obj, log_meta={"run_id": run_id, "call_type": "geo_risk_score"})
|
||
except Exception as _ge:
|
||
logger.warning(f"[Cycle {run_id[:16]}] AI geo risk scoring failed, falling back to algo score: {_ge}")
|
||
geo_score_obj = {**algo_geo_score_obj, "rationale": "Erreur IA — score algorithmique utilisé tel quel.", "top_risks": []}
|
||
geo_score_obj.setdefault("breakdown", algo_geo_score_obj.get("breakdown", {}))
|
||
geo_score_val = int(round(geo_score_obj.get("score") or 0))
|
||
summary["geo_score"] = geo_score_val
|
||
|
||
from services.database import save_geo_risk_snapshot
|
||
save_geo_risk_snapshot(
|
||
run_id=run_id, score=geo_score_obj.get("score", geo_score_val), level=geo_score_obj.get("level", "low"),
|
||
breakdown=geo_score_obj.get("breakdown", {}), top_risks=geo_score_obj.get("top_risks", []),
|
||
rationale=geo_score_obj.get("rationale", ""),
|
||
)
|
||
|
||
# ── Invalidation trigger detection ────────────────────────────────────
|
||
try:
|
||
from services.database import get_custom_patterns as _gcp
|
||
_active_patterns = _gcp()
|
||
_news_headlines = " ".join(
|
||
(n.get("title", "") + " " + (n.get("summary", "") or "")).lower()
|
||
for n in news[:15]
|
||
)
|
||
_triggered = []
|
||
for _pat in _active_patterns:
|
||
_trigger = (_pat.get("invalidation_trigger") or "").lower().strip()
|
||
if not _trigger:
|
||
continue
|
||
# Simple keyword matching: split trigger into words and check majority match
|
||
_words = [w for w in _trigger.split() if len(w) > 3]
|
||
if _words and sum(1 for w in _words if w in _news_headlines) >= max(1, len(_words) // 2):
|
||
_triggered.append({
|
||
"pattern_id": _pat["id"],
|
||
"pattern_name": _pat["name"],
|
||
"trigger": _pat["invalidation_trigger"],
|
||
"probability": _pat.get("invalidation_probability"),
|
||
})
|
||
if _triggered:
|
||
logger.warning(
|
||
f"[Cycle {run_id[:16]}] ⚠ INVALIDATION TRIGGERS FIRED for "
|
||
f"{len(_triggered)} pattern(s): {[t['pattern_name'] for t in _triggered]}"
|
||
)
|
||
summary["invalidation_alerts"] = _triggered
|
||
except Exception as _ie:
|
||
logger.debug(f"[Cycle] Invalidation check failed (non-blocking): {_ie}")
|
||
|
||
quotes = get_all_quotes()
|
||
|
||
# ── Phase 4A: capture price snapshots for scored news ─────────────────
|
||
try:
|
||
from services.price_discovery import capture_price_snapshots as _cap_snap
|
||
_quotes_flat: Dict[str, float] = {}
|
||
for _qs in quotes.values():
|
||
for _q in _qs:
|
||
if _q.get("symbol") and _q.get("price"):
|
||
_quotes_flat[_q["symbol"]] = float(_q["price"])
|
||
_n_snaps = _cap_snap(news, _quotes_flat, run_id)
|
||
if _n_snaps:
|
||
logger.info(f"[Cycle {run_id[:16]}] Phase 4: {_n_snaps} price snapshots captured")
|
||
purge_old_price_snapshots(older_than_days=step_cfg["price_snapshot_retention"]["days"])
|
||
except Exception as _pd_e:
|
||
logger.warning(f"[Cycle] Price snapshot capture failed (non-blocking): {_pd_e}")
|
||
|
||
gauges = get_macro_gauges()
|
||
scenarios = score_macro_scenarios(gauges)
|
||
macro_regime = {"gauges": gauges, "scenarios": scenarios}
|
||
dominant = scenarios.get("dominant", "incertain")
|
||
summary["dominant_regime"] = dominant
|
||
|
||
# ── Step 1.9: Pre-fetch IV context + snapshots (suggestion rules + gate) ─
|
||
iv_context = ""
|
||
_iv_snapshots: Dict[str, Dict] = {}
|
||
try:
|
||
from services.iv_engine import get_iv_context_for_prompt, get_full_iv_snapshot, IV_WATCHLIST
|
||
from services.database import get_mtm_trades_with_traces
|
||
_mtm_pre = get_mtm_trades_with_traces(days=step_cfg["iv_context"]["lookback_days"])
|
||
_trade_tickers_pre = list({
|
||
(t.get("underlying") or "").upper()
|
||
for t in _mtm_pre.get("all_trades", [])
|
||
if t.get("underlying")
|
||
})
|
||
_iv_tickers_pre = (_trade_tickers_pre + IV_WATCHLIST[:6])[:10]
|
||
iv_context = get_iv_context_for_prompt(_iv_tickers_pre)
|
||
# Also collect structured snapshots — used by IV gate before log_trade_entries
|
||
for _tkr in _iv_tickers_pre:
|
||
try:
|
||
_iv_snapshots[_tkr] = get_full_iv_snapshot(_tkr)
|
||
except Exception:
|
||
pass
|
||
if iv_context:
|
||
logger.info(
|
||
f"[Cycle {run_id[:16]}] IV context pre-fetched: {len(_iv_tickers_pre)} tickers, "
|
||
f"{len(_iv_snapshots)} snapshots"
|
||
)
|
||
except Exception as _e:
|
||
logger.warning(f"[Cycle] IV context pre-fetch failed (non-blocking): {_e}")
|
||
|
||
# ── Step 2: Suggest new patterns ──────────────────────────────────────
|
||
logger.info(f"[Cycle {run_id[:16]}] Step 2: suggesting patterns")
|
||
_reliability_map = {}
|
||
try:
|
||
from services.database import get_all_pattern_reliability_map
|
||
_reliability_map = get_all_pattern_reliability_map()
|
||
if _reliability_map:
|
||
logger.info(f"[Cycle {run_id[:16]}] Reliability map: {len(_reliability_map)} patterns")
|
||
except Exception as _re:
|
||
logger.warning(f"[Cycle] Reliability map failed (non-blocking): {_re}")
|
||
# ── Phase 4B: compute price absorption from previous snapshots ───────
|
||
_absorptions = []
|
||
_price_discovery_block = ""
|
||
try:
|
||
from services.price_discovery import compute_absorptions, build_price_discovery_block
|
||
_absorptions = compute_absorptions(
|
||
min_age_minutes=step_cfg["absorption_detection"]["min_age_minutes"],
|
||
max_age_days=step_cfg["absorption_detection"]["max_age_days"],
|
||
)
|
||
_price_discovery_block = build_price_discovery_block(_absorptions)
|
||
opps = sum(1 for a in _absorptions if a["opportunity"])
|
||
if _absorptions:
|
||
logger.info(f"[Cycle {run_id[:16]}] Phase 4: {len(_absorptions)} absorptions, {opps} opportunities")
|
||
except Exception as _abs_e:
|
||
logger.warning(f"[Cycle] Price absorption failed (non-blocking): {_abs_e}")
|
||
|
||
# ── Phase 2: FRED recent macro releases + economic surprise tracking ────
|
||
_fred_releases = []
|
||
_fred_block = ""
|
||
_surprise_block = ""
|
||
try:
|
||
from services.fred_fetcher import (
|
||
get_fred_recent_releases, build_fred_context_block,
|
||
save_fred_releases_to_db, build_economic_surprise_block,
|
||
)
|
||
_fred_key = get_config("fred_api_key") or ""
|
||
_fred_releases = get_fred_recent_releases()
|
||
if _fred_releases:
|
||
logger.info(f"[Cycle {run_id[:16]}] FRED: {len(_fred_releases)} releases fetched")
|
||
# Persist to economic_events with z-score surprise scoring
|
||
try:
|
||
_saved = save_fred_releases_to_db(_fred_releases, api_key=_fred_key)
|
||
if _saved:
|
||
logger.info(f"[Cycle {run_id[:16]}] FRED: {_saved} releases saved to economic_events")
|
||
except Exception as _dbe:
|
||
logger.warning(f"[Cycle] FRED DB save failed (non-blocking): {_dbe}")
|
||
# Add surprise scores to release dicts for richer prompt block
|
||
if _fred_key:
|
||
try:
|
||
from services.database import get_recent_economic_surprises
|
||
_db_surprises = {ev["series_id"]: ev for ev in get_recent_economic_surprises(days=step_cfg["economic_surprises"]["lookback_days"])}
|
||
for rel in _fred_releases:
|
||
sid = rel.get("series_id", "")
|
||
if sid in _db_surprises:
|
||
rel["surprise_zscore"] = _db_surprises[sid].get("surprise_zscore")
|
||
except Exception:
|
||
pass
|
||
_fred_block = build_fred_context_block(_fred_releases)
|
||
_surprise_block = build_economic_surprise_block(days=14)
|
||
if _surprise_block:
|
||
_fred_block = _fred_block + "\n\n" + _surprise_block if _fred_block else _surprise_block
|
||
except Exception as _fe:
|
||
logger.warning(f"[Cycle] FRED fetch failed (non-blocking): {_fe}")
|
||
|
||
try:
|
||
from services.data_fetcher import get_economic_calendar
|
||
calendar = get_economic_calendar()
|
||
# Apply decay to news before suggestion (adds decayed_score + age_hours)
|
||
from services.ai_analyzer import apply_news_decay as _apply_decay
|
||
news = _apply_decay(news)
|
||
|
||
# ── Tech indicators for top tickers ──────────────────────────────
|
||
_tech_block = ""
|
||
try:
|
||
_ti_enabled = (get_config("tech_indicators_enabled") or "true").lower() == "true"
|
||
if _ti_enabled:
|
||
from services.technical_indicators import compute_indicators, format_indicators_for_prompt
|
||
_ti_list = [s.strip() for s in (get_config("tech_indicators_list") or "rsi,ma,bollinger,atr").split(",") if s.strip()]
|
||
_horizon_days = 45 # default mid-range if no trade horizon context yet
|
||
_ti_lines = []
|
||
# Pick up to 5 tickers from quotes (one per asset class)
|
||
_seen_tickers: set = set()
|
||
for _cls, _qs in quotes.items():
|
||
for _q in _qs[:1]:
|
||
_sym = _q.get("symbol", "")
|
||
if _sym and _sym not in _seen_tickers:
|
||
_seen_tickers.add(_sym)
|
||
_ind = compute_indicators(_sym, _horizon_days, enabled_indicators=_ti_list)
|
||
_blk = format_indicators_for_prompt(_ind)
|
||
if _blk:
|
||
_ti_lines.append(_blk)
|
||
if len(_seen_tickers) >= 5:
|
||
break
|
||
if len(_seen_tickers) >= 5:
|
||
break
|
||
if _ti_lines:
|
||
_tech_block = "\n".join(_ti_lines)
|
||
except Exception as _te:
|
||
logger.warning(f"[Cycle] Tech indicators failed (non-blocking): {_te}")
|
||
|
||
# ── Portfolio context (open positions + recent moves) ──────────
|
||
_portfolio_block = ""
|
||
_open_trades_snap = []
|
||
_portfolio_conc = {}
|
||
try:
|
||
_open_trades_snap = get_open_trades_with_moves()
|
||
_portfolio_conc = get_portfolio_concentration(_open_trades_snap)
|
||
_portfolio_block = build_portfolio_context_block(_open_trades_snap, _portfolio_conc)
|
||
logger.info(f"[Cycle {run_id[:16]}] Portfolio context: {len(_open_trades_snap)} open trades")
|
||
except Exception as _pe:
|
||
logger.warning(f"[Cycle] Portfolio context failed (non-blocking): {_pe}")
|
||
|
||
# ── Save full context snapshot ─────────────────────────────────
|
||
try:
|
||
from services.ai_analyzer import apply_news_decay as _apply_decay2, partition_news_by_age as _part
|
||
_news_snap = _part(_apply_decay2(news), cycle_meta.get("delta_minutes", 180))
|
||
_context_snapshot = {
|
||
"cycle_meta": cycle_meta,
|
||
"macro_regime": {
|
||
"dominant": (macro_regime or {}).get("scenarios", {}).get("dominant"),
|
||
"scores": (macro_regime or {}).get("scenarios", {}).get("scores"),
|
||
"gauges_summary": {k: v.get("value") for k, v in ((macro_regime or {}).get("gauges", {})).items()},
|
||
},
|
||
"geo_score": geo_score_obj,
|
||
"news_partitioned": {
|
||
"inter_cycle": [{"title": n.get("title"), "source": n.get("source"), "decayed_score": n.get("decayed_score"), "age_hours": n.get("age_hours")} for n in _news_snap["inter_cycle"][:10]],
|
||
"recent_24h": [{"title": n.get("title"), "source": n.get("source"), "decayed_score": n.get("decayed_score")} for n in _news_snap["recent_24h"][:10]],
|
||
"older": [{"title": n.get("title"), "source": n.get("source")} for n in _news_snap["older"][:5]],
|
||
},
|
||
"fred_releases": _fred_releases,
|
||
"price_discovery": [
|
||
{k: v for k, v in a.items() if k != "article_hash"}
|
||
for a in _absorptions[:10]
|
||
],
|
||
"tech_indicators_block": _tech_block,
|
||
"iv_context_preview": iv_context[:500] if iv_context else "",
|
||
"calendar": calendar[:8] if calendar else [],
|
||
"quotes_summary": {cls: [{"symbol": q.get("symbol"), "price": q.get("price"), "change_pct": q.get("change_pct")} for q in qs[:3]] for cls, qs in quotes.items()},
|
||
"portfolio_open_positions": {
|
||
"count": len(_open_trades_snap),
|
||
"concentration": _portfolio_conc,
|
||
"trades": [
|
||
{k: v for k, v in t.items() if k != "id"}
|
||
for t in _open_trades_snap
|
||
],
|
||
},
|
||
}
|
||
save_cycle_context_snapshot(run_id, _context_snapshot)
|
||
logger.info(f"[Cycle {run_id[:16]}] Context snapshot saved")
|
||
except Exception as _snap_e:
|
||
logger.warning(f"[Cycle] Context snapshot save failed (non-blocking): {_snap_e}")
|
||
|
||
# ── Build convergence block from previous cycle's scored patterns ──
|
||
_convergence_block_for_suggest = ""
|
||
try:
|
||
from services.database import get_patterns_with_last_score
|
||
from services.ai_analyzer import _compute_convergence
|
||
_prev_scored = get_patterns_with_last_score()
|
||
_prev_with_scores = [
|
||
{"pattern_id": p.get("id"), "score": p.get("last_score") or 0,
|
||
"recommended_trade": {"underlying": (p.get("suggested_trades") or [{}])[0].get("underlying", "")} if p.get("suggested_trades") else {},
|
||
"trade_rankings": [{"underlying": t.get("underlying", "")} for t in (p.get("suggested_trades") or [])],
|
||
"geo_trigger": p.get("name", "")}
|
||
for p in _prev_scored if p.get("last_score") is not None
|
||
]
|
||
if _prev_with_scores:
|
||
_, _convergence_block_for_suggest = _compute_convergence(_prev_with_scores, _prev_scored)
|
||
if _convergence_block_for_suggest:
|
||
logger.info(f"[Cycle {run_id[:16]}] Convergence block built from {len(_prev_with_scores)} previously scored patterns")
|
||
except Exception as _cbe:
|
||
logger.warning(f"[Cycle] Convergence block build failed (non-blocking): {_cbe}")
|
||
|
||
# ── Build institutional reports block ──────────────────────────
|
||
_institutional_block = ""
|
||
try:
|
||
from services.ai_analyzer import build_institutional_block
|
||
_institutional_block = build_institutional_block(days=step_cfg["institutional_block"]["lookback_days"])
|
||
if _institutional_block:
|
||
logger.info(f"[Cycle {run_id[:16]}] Institutional block injected")
|
||
except Exception as _ibe:
|
||
logger.warning(f"[Cycle] Institutional block failed (non-blocking): {_ibe}")
|
||
|
||
# ── COT + Forward Curves (background refresh, non-blocking) ───────────────
|
||
try:
|
||
from services.cot_fetcher import fetch_all_cot
|
||
from services.forward_curve import fetch_forward_curves
|
||
from services.database import save_cot_data, save_forward_curves as _sfc
|
||
cot_data = fetch_all_cot()
|
||
if cot_data:
|
||
save_cot_data(cot_data)
|
||
curve_data = fetch_forward_curves()
|
||
if curve_data:
|
||
_sfc(curve_data)
|
||
except Exception as _e:
|
||
import logging as _logging
|
||
_logging.getLogger(__name__).warning(f"COT/Curve refresh failed (non-blocking): {_e}")
|
||
|
||
suggestions = suggest_patterns_from_market_context(
|
||
news, quotes, calendar, macro_regime=macro_regime, geo_score=geo_score_obj,
|
||
portfolio_lessons=portfolio_lessons,
|
||
reliability_map=_reliability_map or None,
|
||
iv_context=iv_context,
|
||
cycle_meta=cycle_meta,
|
||
tech_indicators_block=_tech_block,
|
||
fred_block=_fred_block,
|
||
price_discovery_block=_price_discovery_block,
|
||
portfolio_context_block=_portfolio_block,
|
||
convergence_block=_convergence_block_for_suggest,
|
||
run_id=run_id,
|
||
)
|
||
except Exception as e:
|
||
logger.warning(f"[Cycle] Suggestion step failed: {e}")
|
||
suggestions = []
|
||
|
||
summary["patterns_suggested"] = len(suggestions)
|
||
logger.info(f"[Cycle {run_id[:16]}] Suggested {len(suggestions)} patterns from AI")
|
||
|
||
# ── Step 3: Filter by similarity (Embedding cosinus ou Jaccard fallback) ─
|
||
existing = get_custom_patterns()
|
||
logger.info(f"[Cycle {run_id[:16]}] Step 3: {len(existing)} existing patterns, threshold={sim_threshold}")
|
||
added_count = 0
|
||
for s in suggestions:
|
||
if not _is_duplicate_pattern(s, existing, ai_key, jaccard_threshold=sim_threshold, embed_threshold=step_cfg["duplicate_detection"]["embedding_threshold"]):
|
||
|
||
# Capture returned ID so the pattern has a valid id for scoring
|
||
assigned_id = save_custom_pattern(s)
|
||
s["id"] = assigned_id
|
||
existing.append(s)
|
||
added_count += 1
|
||
logger.info(f"[Cycle] Added pattern '{s.get('name')}' id={assigned_id}")
|
||
# ── Save suggestion reasoning trace ───────────────────────────
|
||
_top_news_ctx = [
|
||
{"title": n.get("title", "")[:120], "impact": round(float(n.get("impact_score") or 0), 2), "source": n.get("source", "")}
|
||
for n in sorted(news, key=lambda x: -(float(x.get("impact_score") or 0)))[:8]
|
||
]
|
||
save_reasoning_trace(
|
||
run_id=run_id,
|
||
trace_type="suggestion",
|
||
pattern_id=assigned_id,
|
||
input_context={
|
||
"geo_score": geo_score_val,
|
||
"macro_dominant": dominant,
|
||
"macro_scores": scenarios.get("scores", {}),
|
||
"top_news": _top_news_ctx,
|
||
"cycle_run_id": run_id,
|
||
},
|
||
output={
|
||
"name": s.get("name"),
|
||
"description": s.get("description"),
|
||
"macro_fit": s.get("macro_fit"),
|
||
"expected_move_pct": s.get("expected_move_pct"),
|
||
"probability": s.get("probability"),
|
||
"horizon_days": s.get("horizon_days"),
|
||
"suggested_trades": s.get("suggested_trades", []),
|
||
"keywords": s.get("keywords", []),
|
||
"triggers": s.get("triggers", []),
|
||
},
|
||
reasoning_summary=(s.get("macro_fit") or s.get("description") or "")[:300],
|
||
geo_score=geo_score_val,
|
||
macro_dominant=dominant,
|
||
)
|
||
else:
|
||
logger.debug(f"[Cycle] Filtered '{s.get('name')}' — doublon détecté")
|
||
|
||
summary["patterns_added"] = added_count
|
||
|
||
# ── Step 3.1: Classify unclassified patterns (category + signal_direction) ─
|
||
try:
|
||
from services.database import get_unclassified_patterns, update_pattern_classification
|
||
from services.ai_analyzer import classify_patterns_batch
|
||
_unclassified = get_unclassified_patterns()
|
||
if _unclassified:
|
||
logger.info(f"[Cycle {run_id[:16]}] Classifying {len(_unclassified)} unclassified patterns")
|
||
_classifications = classify_patterns_batch(_unclassified)
|
||
for _cl in _classifications:
|
||
if _cl.get("id") and _cl.get("category"):
|
||
update_pattern_classification(_cl["id"], _cl["category"], _cl.get("signal_direction", "neutral"))
|
||
logger.info(f"[Cycle {run_id[:16]}] Classified {len(_classifications)} patterns")
|
||
except Exception as _cle:
|
||
logger.warning(f"[Cycle] Pattern classification failed (non-blocking): {_cle}")
|
||
|
||
# ── Step 3.5: Collect risk cluster context ───────────────────────────
|
||
risk_cluster_context = ""
|
||
try:
|
||
from services.database import get_risk_clusters as _grc
|
||
_clusters = _grc()
|
||
risk_cluster_context = _clusters.get("risk_prompt_context", "")
|
||
if risk_cluster_context:
|
||
logger.info(
|
||
f"[Cycle {run_id[:16]}] Risk clusters: "
|
||
f"{len(_clusters.get('clusters', []))} facteurs, "
|
||
f"saturés={_clusters.get('saturated_factors', [])}"
|
||
)
|
||
except Exception as _re:
|
||
logger.warning(f"[Cycle] Risk cluster context failed (non-blocking): {_re}")
|
||
|
||
# ── Step 3.6: Refresh IV context (with full trade+watchlist scope) ──────
|
||
try:
|
||
from services.iv_engine import get_iv_context_for_prompt, IV_WATCHLIST
|
||
# Collect underlyings from current trade journal + default watchlist
|
||
from services.database import get_mtm_trades_with_traces
|
||
_mtm = get_mtm_trades_with_traces(days=step_cfg["iv_context"]["lookback_days"])
|
||
_trade_tickers = list({
|
||
(t.get("underlying") or "").upper()
|
||
for t in _mtm.get("all_trades", [])
|
||
if t.get("underlying")
|
||
})
|
||
_iv_tickers = (_trade_tickers + IV_WATCHLIST[:6])[:10]
|
||
iv_context = get_iv_context_for_prompt(_iv_tickers)
|
||
if iv_context:
|
||
logger.info(f"[Cycle {run_id[:16]}] IV context collected for {len(_iv_tickers)} tickers")
|
||
# Invalidate the options_vol watchlist cache so the frontend sees fresh data
|
||
try:
|
||
from routers.options_vol import _iv_cache
|
||
stale_keys = [k for k in list(_iv_cache.keys()) if k.startswith("watchlist:") or k.startswith("snapshot:")]
|
||
for k in stale_keys:
|
||
_iv_cache.pop(k, None)
|
||
if stale_keys:
|
||
logger.info(f"[Cycle] IV cache invalidated ({len(stale_keys)} entries)")
|
||
except Exception:
|
||
pass
|
||
except Exception as _e:
|
||
logger.warning(f"[Cycle] IV context collection failed (non-blocking): {_e}")
|
||
|
||
# ── Step 4: Score ALL patterns ────────────────────────────────────────
|
||
# Verify all patterns have IDs before scoring (guard against stale data)
|
||
patterns_with_id = [p for p in existing if p.get("id")]
|
||
patterns_without_id = [p.get("name", "?") for p in existing if not p.get("id")]
|
||
if patterns_without_id:
|
||
logger.warning(f"[Cycle] {len(patterns_without_id)} patterns have no id, skipping: {patterns_without_id}")
|
||
|
||
logger.info(f"[Cycle {run_id[:16]}] Step 4: scoring {len(patterns_with_id)} patterns (of {len(existing)} total)")
|
||
from services.database import get_analysis_config
|
||
template = get_analysis_config().get("template") or DEFAULT_ANALYSIS_TEMPLATE
|
||
try:
|
||
scored = score_patterns_with_context(
|
||
patterns=patterns_with_id,
|
||
recent_news=news,
|
||
quotes_by_class=quotes,
|
||
geo_score=geo_score_obj,
|
||
template=template,
|
||
top_n=len(patterns_with_id),
|
||
category_filter=None,
|
||
macro_regime=macro_regime,
|
||
portfolio_lessons=portfolio_lessons,
|
||
iv_context=iv_context,
|
||
risk_context=risk_cluster_context,
|
||
cycle_meta=cycle_meta,
|
||
tech_indicators_block=_tech_block,
|
||
fred_block=_fred_block,
|
||
price_discovery_block=_price_discovery_block,
|
||
portfolio_context_block=_portfolio_block,
|
||
institutional_block=_institutional_block,
|
||
run_id=run_id,
|
||
)
|
||
scored_with_id = [s for s in scored if s.get("pattern_id")]
|
||
scored_without_id = [s for s in scored if not s.get("pattern_id")]
|
||
if scored_without_id:
|
||
logger.warning(f"[Cycle] {len(scored_without_id)} scored results have no pattern_id — they will NOT be saved to history")
|
||
logger.info(f"[Cycle {run_id[:16]}] Scoring returned {len(scored)} results ({len(scored_with_id)} with valid id)")
|
||
except Exception as e:
|
||
logger.error(f"[Cycle] Scoring failed: {e}", exc_info=True)
|
||
scored = []
|
||
|
||
summary["patterns_scored"] = len(scored)
|
||
|
||
# ── Enrich scored patterns with original data not in GPT-4o response ─
|
||
# GPT-4o scoring doesn't return expected_move_pct or suggested_trades —
|
||
# copy from the original pattern so log_trade_entries can use them.
|
||
_pmap = {p.get("id"): p for p in patterns_with_id}
|
||
for s in scored:
|
||
orig = _pmap.get(s.get("pattern_id", ""), {})
|
||
if orig:
|
||
if not s.get("expected_move_pct") and orig.get("expected_move_pct"):
|
||
s["expected_move_pct"] = orig["expected_move_pct"]
|
||
logger.debug(f"[Cycle] Enriched '{orig.get('name')}' expected_move_pct={orig['expected_move_pct']}")
|
||
if not s.get("trade_rankings") and not s.get("suggested_trades"):
|
||
s["suggested_trades"] = orig.get("suggested_trades", [])
|
||
# Propagate category + signal_direction to scored result for display
|
||
if orig.get("category") and not s.get("category"):
|
||
s["category"] = orig["category"]
|
||
if orig.get("signal_direction") and not s.get("signal_direction"):
|
||
s["signal_direction"] = orig["signal_direction"]
|
||
|
||
# ── Log convergence summary ───────────────────────────────────────────
|
||
try:
|
||
_top_conv = [s for s in scored if s.get("convergence_count", 0) > 0]
|
||
if _top_conv:
|
||
logger.info(
|
||
f"[Cycle {run_id[:16]}] Convergence: {len(_top_conv)} patterns with cross-pattern boost "
|
||
f"— top: {_top_conv[0].get('convergence_underlying','?')} ×{_top_conv[0].get('convergence_count',0)+1}"
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
# ── Step 5: Log everything ────────────────────────────────────────────
|
||
logger.info(f"[Cycle {run_id[:16]}] Step 5: logging")
|
||
scoring_run_id = save_pattern_scores(scored, meta={
|
||
"geo_score": geo_score_val,
|
||
"total": len(scored),
|
||
"cycle_run_id": run_id,
|
||
"trigger": trigger,
|
||
})
|
||
|
||
# ── Save scoring reasoning traces (one per scored pattern) ────────────
|
||
_macro_scores_ctx = scenarios.get("scores", {})
|
||
_asset_bias_ctx = scenarios.get("asset_bias", {}).get(dominant, {})
|
||
for sp in scored:
|
||
pid = sp.get("pattern_id", "")
|
||
if not pid:
|
||
continue
|
||
orig = _pmap.get(pid, {})
|
||
save_reasoning_trace(
|
||
run_id=scoring_run_id,
|
||
trace_type="scoring",
|
||
pattern_id=pid,
|
||
input_context={
|
||
"geo_score": geo_score_val,
|
||
"macro_dominant": dominant,
|
||
"macro_scores": _macro_scores_ctx,
|
||
"asset_class": sp.get("asset_class") or orig.get("asset_class"),
|
||
"asset_bias": _asset_bias_ctx.get(sp.get("asset_class") or orig.get("asset_class", ""), "neutral"),
|
||
"expected_move_pct": sp.get("expected_move_pct") or orig.get("expected_move_pct"),
|
||
"cycle_run_id": run_id,
|
||
},
|
||
output={
|
||
"score": sp.get("score"),
|
||
"confidence": sp.get("confidence"),
|
||
"buckets": sp.get("buckets", []),
|
||
"key_catalyst": sp.get("key_catalyst"),
|
||
"summary": sp.get("summary"),
|
||
"trade_rankings": sp.get("trade_rankings", []),
|
||
"recommended_trade": sp.get("recommended_trade", {}),
|
||
"has_strong_contra": sp.get("has_strong_contra", False),
|
||
"geo_trigger": sp.get("geo_trigger"),
|
||
},
|
||
reasoning_summary=((sp.get("key_catalyst") or "") + " | " + (sp.get("summary") or ""))[:400],
|
||
geo_score=geo_score_val,
|
||
macro_dominant=dominant,
|
||
)
|
||
logger.info(f"[Cycle {run_id[:16]}] Saved {len([s for s in scored if s.get('pattern_id')])} reasoning traces")
|
||
|
||
top_patterns_log = sorted(
|
||
[{"pattern_id": sp.get("pattern_id"), "name": sp.get("geo_trigger"), "score": sp.get("score", 0)}
|
||
for sp in scored if sp.get("score", 0) > 0],
|
||
key=lambda x: -x["score"]
|
||
)[:10]
|
||
|
||
log_geo_alert(geo_score=geo_score_val, top_patterns=top_patterns_log,
|
||
news_count=len(news), run_id=scoring_run_id)
|
||
|
||
_options_assessment = None
|
||
|
||
# ── IV Gate: block ALERT-verdict trades before logging ────────────────
|
||
_iv_gate_blocked: List[Dict] = []
|
||
try:
|
||
from services.database import get_config as _gc
|
||
if (_gc("iv_gate_enabled") or "true").lower() == "true":
|
||
# Augment _iv_snapshots with any new tickers from scored that weren't in the pre-fetch
|
||
from services.iv_engine import get_full_iv_snapshot as _gfiv
|
||
_scored_tickers = {
|
||
(tr.get("underlying") or "").upper()
|
||
for sp in scored
|
||
for tr in (sp.get("trade_rankings") or sp.get("suggested_trades") or [])
|
||
if tr.get("underlying")
|
||
}
|
||
for _stk in _scored_tickers - set(_iv_snapshots.keys()):
|
||
try:
|
||
_iv_snapshots[_stk] = _gfiv(_stk)
|
||
except Exception:
|
||
pass
|
||
|
||
scored, _iv_gate_blocked = _apply_iv_gate(scored, _iv_snapshots)
|
||
if _iv_gate_blocked:
|
||
summary["iv_gate_blocked"] = len(_iv_gate_blocked)
|
||
logger.warning(
|
||
f"[IVGate] {len(_iv_gate_blocked)} trade(s) blocked by IV rules: "
|
||
+ ", ".join(f"{b['ticker']} {b['strategy']}" for b in _iv_gate_blocked)
|
||
)
|
||
for _blk in _iv_gate_blocked:
|
||
try:
|
||
from services.database import log_skipped_trade
|
||
log_skipped_trade(
|
||
run_id=scoring_run_id,
|
||
pattern_id=_blk.get("pattern_id", ""),
|
||
pattern_name=_blk.get("pattern_name", ""),
|
||
underlying=_blk.get("ticker", ""),
|
||
strategy=_blk.get("strategy", ""),
|
||
score=_blk.get("score", 0),
|
||
expected_move_pct=0,
|
||
skip_detail=(
|
||
f"[IV_GATE] ALERT — {'; '.join(_blk.get('iv_issues', [])[:2])}. "
|
||
f"Stratégie optimale: {_blk.get('optimal_strategy', '?')}"
|
||
),
|
||
asset_class=_blk.get("asset_class", ""),
|
||
)
|
||
except Exception:
|
||
pass
|
||
except Exception as _ige:
|
||
logger.warning(f"[IVGate] Failed (non-blocking): {_ige}")
|
||
|
||
log_trade_entries(run_id=scoring_run_id, scored_patterns=scored, quotes=quotes)
|
||
|
||
# ── Step 5.2: Options Technical Agent — validate newly logged trades ──
|
||
try:
|
||
from services.options_technical_agent import assess_logged_trades, save_assessments_to_db
|
||
_options_assessment = assess_logged_trades(
|
||
scoring_run_id=scoring_run_id,
|
||
scored=scored,
|
||
ai_key=ai_key,
|
||
)
|
||
if _options_assessment and _options_assessment.get("assessments"):
|
||
save_assessments_to_db(_options_assessment["assessments"], scoring_run_id)
|
||
summary["options_assessment"] = {
|
||
"n_ok": _options_assessment.get("n_ok", 0),
|
||
"n_warn": _options_assessment.get("n_warn", 0),
|
||
"n_alert": _options_assessment.get("n_alert", 0),
|
||
"global_score": _options_assessment.get("global_score"),
|
||
}
|
||
logger.info(
|
||
f"[OptionsTech] Assessment done — OK={_options_assessment.get('n_ok')} "
|
||
f"WARN={_options_assessment.get('n_warn')} ALERT={_options_assessment.get('n_alert')}"
|
||
)
|
||
except Exception as _ota:
|
||
logger.warning(f"[OptionsTech] Agent failed (non-blocking): {_ota}")
|
||
|
||
# ── Step 5.1: Portfolio monitor — conflict & concentration check ──────
|
||
try:
|
||
from services.portfolio_risk import analyze_simulation_portfolio
|
||
_risk = analyze_simulation_portfolio()
|
||
if not step_cfg["portfolio_monitor"]["enabled"]:
|
||
logger.info("[PortfolioMonitor] Disabled via cycle_step_config — skipping AI monitor")
|
||
elif _risk.get("alerts"):
|
||
logger.info(f"[PortfolioMonitor] {len(_risk['alerts'])} alerts ({len(_risk['conflicts'])} conflicts) — running AI monitor")
|
||
_pm = _run_portfolio_monitor(_risk, scoring_run_id)
|
||
if _pm:
|
||
summary["portfolio_monitor"] = {
|
||
"alerts": len(_risk["alerts"]),
|
||
"conflicts": len(_risk["conflicts"]),
|
||
"assessment": _pm.get("assessment", ""),
|
||
"actions_count": len(_pm.get("actions", [])),
|
||
}
|
||
else:
|
||
logger.info(f"[PortfolioMonitor] OK — {_risk['open_count']} positions, no alerts")
|
||
except Exception as _pme:
|
||
logger.warning(f"[PortfolioMonitor] Failed (non-blocking): {_pme}")
|
||
|
||
# Auto-add any new underlying tickers to the IV watchlist
|
||
if not step_cfg["watchlist_auto_add"]["enabled"]:
|
||
logger.info("[Watchlist] Auto-add disabled via cycle_step_config — skipping")
|
||
else:
|
||
try:
|
||
from services.database import _normalize_ticker, add_watchlist_ticker, get_watchlist_tickers
|
||
from services.iv_engine import _resolve_ticker, bootstrap_iv_history
|
||
existing = set(get_watchlist_tickers())
|
||
new_proxies = set()
|
||
for sp in scored:
|
||
for trade in (sp.get("trade_rankings") or sp.get("suggested_trades") or []):
|
||
underlying = trade.get("underlying") or sp.get("underlying") or ""
|
||
if underlying:
|
||
normalized = _normalize_ticker(underlying.upper()) # WHEAT→ZW=F, EUR/USD→EURUSD=X
|
||
proxy = _resolve_ticker(normalized) # ZW=F→WEAT, EURUSD=X→FXE
|
||
if proxy not in existing:
|
||
new_proxies.add(proxy)
|
||
for proxy in new_proxies:
|
||
if add_watchlist_ticker(proxy, added_by="cycle"):
|
||
logger.info(f"[Watchlist] Auto-added new ticker: {proxy}")
|
||
log_system_event("INFO", "auto_cycle", f"Nouveau ticker ajouté à la watchlist IV: {proxy}", cycle_id=scoring_run_id, ticker=proxy)
|
||
if new_proxies:
|
||
bootstrap_iv_history(tickers=list(new_proxies), min_existing=0)
|
||
except Exception as _we:
|
||
logger.warning(f"[Watchlist] Auto-add failed: {_we}")
|
||
|
||
gauges_summary = {
|
||
k: {"value": v.get("value"), "change_pct": v.get("change_pct"), "label": v.get("label")}
|
||
for k, v in gauges.items()
|
||
if v.get("value") is not None or v.get("change_pct") is not None
|
||
}
|
||
log_macro_regime(dominant=dominant, scores=scenarios.get("scores", {}),
|
||
reasons=scenarios.get("reasons", {}), gauges_summary=gauges_summary)
|
||
|
||
# Update macro cache so the UI sees fresh data immediately
|
||
from routers.market_data import _macro_cache # type: ignore
|
||
import datetime as _dt
|
||
_macro_cache["data"] = {"gauges": gauges, "scenarios": scenarios,
|
||
"fetched_at": datetime.utcnow().isoformat(), "cached": False}
|
||
_macro_cache["ts"] = _dt.datetime.utcnow()
|
||
|
||
# ── Step 5.5: Bayesian posterior update (Sprint 4.1) ─────────────────
|
||
try:
|
||
from services.database import update_bayesian_posteriors
|
||
_n_updated = update_bayesian_posteriors()
|
||
if _n_updated:
|
||
logger.info(f"[Cycle {run_id[:16]}] Bayesian posteriors mis à jour : {_n_updated} patterns")
|
||
except Exception as _be:
|
||
logger.warning(f"[Cycle] Bayesian update failed (non-blocking): {_be}")
|
||
|
||
# ── Step 5.6: Régime clustering (Sprint 4.2) ──────────────────────────
|
||
if not step_cfg["regime_clustering"]["enabled"]:
|
||
logger.info("[Cycle] Régime clustering disabled via cycle_step_config — skipping")
|
||
else:
|
||
try:
|
||
from services.database import detect_and_save_regime_clusters
|
||
_cluster_result = detect_and_save_regime_clusters(
|
||
n_clusters=step_cfg["regime_clustering"]["n_clusters"],
|
||
days=step_cfg["regime_clustering"]["lookback_days"],
|
||
)
|
||
if "current_cluster" in _cluster_result:
|
||
logger.info(
|
||
f"[Cycle {run_id[:16]}] Régime cluster : {_cluster_result.get('current_label')} "
|
||
f"(cluster {_cluster_result.get('current_cluster')}, "
|
||
f"anomalie={_cluster_result.get('current_anomaly')})"
|
||
)
|
||
except Exception as _ce:
|
||
logger.warning(f"[Cycle] Régime clustering failed (non-blocking): {_ce}")
|
||
|
||
# ── Step 5.7: Wavelet signal scan (watchlist instruments) ────────────
|
||
_wavelet_results: list = []
|
||
try:
|
||
from services.wavelet_signals import compute_and_save_wavelet_signals
|
||
_wavelet_results = compute_and_save_wavelet_signals(run_id)
|
||
summary["wavelet_signals_count"] = len(_wavelet_results)
|
||
logger.info(f"[Cycle {run_id[:16]}] Wavelet signals: {len(_wavelet_results)} detected")
|
||
except Exception as _we:
|
||
logger.warning(f"[Cycle] Wavelet signal scan failed (non-blocking): {_we}")
|
||
summary["wavelet_signals_count"] = 0
|
||
|
||
# ── Step 6: GPT-4o cycle commentary ──────────────────────────────────
|
||
logger.info(f"[Cycle {run_id[:16]}] Step 6: generating commentary")
|
||
commentary = _generate_cycle_commentary(
|
||
scored=scored, dominant=dominant, scenarios=scenarios,
|
||
geo_score_val=geo_score_val, news=news, gauges=gauges,
|
||
)
|
||
# Attach lessons metadata to commentary so UI can display it
|
||
if commentary and portfolio_lessons:
|
||
try:
|
||
import json as _json
|
||
c = _json.loads(commentary) if isinstance(commentary, str) else commentary
|
||
c["lessons_from_report"] = portfolio_lessons.get("created_at", "")[:16].replace("T", " ")
|
||
c["lessons_headline"] = portfolio_lessons.get("headline", "")[:100]
|
||
commentary = _json.dumps(c, ensure_ascii=False)
|
||
except Exception:
|
||
pass
|
||
summary["commentary"] = commentary
|
||
|
||
# ── Finalize ──────────────────────────────────────────────────────────
|
||
summary["status"] = "completed"
|
||
update_cycle_run(
|
||
run_id,
|
||
completed_at=datetime.utcnow().isoformat(),
|
||
patterns_suggested=summary["patterns_suggested"],
|
||
patterns_added=summary["patterns_added"],
|
||
patterns_scored=summary["patterns_scored"],
|
||
geo_score=geo_score_val,
|
||
dominant_regime=dominant,
|
||
commentary=commentary,
|
||
status="completed",
|
||
wavelet_signals_count=summary.get("wavelet_signals_count", 0),
|
||
)
|
||
_current_status["last_run_at"] = datetime.utcnow().isoformat()
|
||
logger.info(f"[Cycle {run_id[:16]}] Completed — {added_count} new patterns, {len(scored)} scored")
|
||
|
||
# ── Step 6.1: Institutional report absorption tracking ───────────────
|
||
try:
|
||
from services.database import get_conn as _get_conn
|
||
import json as _json
|
||
_commentary_text = ""
|
||
if commentary:
|
||
try:
|
||
_c = _json.loads(commentary) if isinstance(commentary, str) else commentary
|
||
_commentary_text = " ".join([
|
||
str(_c.get("narrative", "")),
|
||
str(_c.get("key_catalyst", "")),
|
||
" ".join(str(x) for x in (_c.get("risks", []) or [])),
|
||
]).lower()
|
||
except Exception:
|
||
_commentary_text = str(commentary).lower()
|
||
|
||
if _commentary_text:
|
||
_conn = _get_conn()
|
||
try:
|
||
cutoff = (datetime.utcnow() - timedelta(days=step_cfg["institutional_absorption"]["lookback_days"])).strftime("%Y-%m-%d")
|
||
_inst_rows = _conn.execute(
|
||
"SELECT id, key_points_json FROM institutional_reports "
|
||
"WHERE report_date >= ? AND (absorbed_score IS NULL OR absorbed_score = 0)",
|
||
(cutoff,),
|
||
).fetchall()
|
||
for _ir in _inst_rows:
|
||
try:
|
||
_kps = _json.loads(_ir["key_points_json"] or "[]")
|
||
_kp_words = set(
|
||
w for kp in _kps for w in kp.lower().split()
|
||
if len(w) > 4
|
||
)
|
||
if not _kp_words:
|
||
continue
|
||
_overlap = sum(1 for w in _kp_words if w in _commentary_text)
|
||
_score = min(1.0, _overlap / max(len(_kp_words), 1))
|
||
if _score > 0:
|
||
_conn.execute(
|
||
"UPDATE institutional_reports SET absorbed_score=?, injected_in_cycle_id=? WHERE id=?",
|
||
(round(_score, 3), run_id, _ir["id"]),
|
||
)
|
||
except Exception:
|
||
pass
|
||
_conn.commit()
|
||
logger.info(f"[Cycle {run_id[:16]}] Absorption tracked for {len(_inst_rows)} institutional reports")
|
||
finally:
|
||
_conn.close()
|
||
except Exception as _abs_e:
|
||
logger.warning(f"[Cycle] Absorption tracking failed (non-blocking): {_abs_e}")
|
||
|
||
# ── Step 6.5: Generate full cycle report ─────────────────────────────
|
||
try:
|
||
_cycle_report = _generate_cycle_report(
|
||
run_id=run_id,
|
||
scored=scored,
|
||
dominant=dominant,
|
||
scenarios=scenarios,
|
||
geo_score_val=geo_score_val,
|
||
news=news,
|
||
gauges=gauges,
|
||
ai_key=ai_key,
|
||
added_patterns=[s for s in suggestions if s.get("id")],
|
||
scoring_run_id=scoring_run_id,
|
||
portfolio_monitor=summary.get("portfolio_monitor"),
|
||
commentary=commentary,
|
||
options_assessment=_options_assessment,
|
||
wavelet_signals=_wavelet_results,
|
||
)
|
||
if _cycle_report:
|
||
from services.database import save_cycle_report
|
||
save_cycle_report(run_id, _cycle_report)
|
||
logger.info(f"[Cycle {run_id[:16]}] Cycle report saved")
|
||
except Exception as _rpe:
|
||
logger.warning(f"[Cycle] Cycle report failed (non-blocking): {_rpe}")
|
||
|
||
# ── Step 7: Auto portfolio snapshot ──────────────────────────────────
|
||
# Generate (or refresh) the portfolio report so the NEXT cycle has
|
||
# fresh performance lessons. Runs in background to not block the cycle.
|
||
import threading
|
||
threading.Thread(
|
||
target=_auto_portfolio_snapshot,
|
||
args=(ai_key,),
|
||
daemon=True,
|
||
name=f"portfolio-snapshot-{run_id[:8]}",
|
||
).start()
|
||
|
||
except Exception as e:
|
||
logger.error(f"[Cycle {run_id[:16]}] Fatal error: {e}", exc_info=True)
|
||
try:
|
||
from services.database import update_cycle_run
|
||
update_cycle_run(run_id, status="error", completed_at=datetime.utcnow().isoformat())
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
_current_status["running"] = False
|
||
_cycle_lock.release()
|
||
|
||
return summary
|
||
|
||
|
||
def _generate_cycle_commentary(
|
||
scored: List[Dict], dominant: str, scenarios: Dict,
|
||
geo_score_val: int, news: List[Dict], gauges: Dict,
|
||
) -> Optional[str]:
|
||
"""Ask GPT-4o to explain current performance of top/bottom patterns."""
|
||
try:
|
||
from services.database import get_trade_entry_prices
|
||
from services.ai_analyzer import _chat
|
||
|
||
# Get recent trade P&L for context
|
||
entries = get_trade_entry_prices(get_cycle_step_config()["cycle_commentary"]["lookback_days"])
|
||
trade_summary = []
|
||
for e in entries[:15]:
|
||
trade_summary.append({
|
||
"pattern": e.get("pattern_name", ""),
|
||
"underlying": e.get("underlying", ""),
|
||
"strategy": e.get("strategy", ""),
|
||
"score_at_entry": e.get("score_at_entry", 0),
|
||
"entry_date": e.get("entry_date", ""),
|
||
})
|
||
|
||
# Top 5 scored patterns now
|
||
top_scored = sorted(scored, key=lambda x: -(x.get("score") or 0))[:5]
|
||
top_scored_summary = [
|
||
{"name": s.get("geo_trigger"), "score": s.get("score"), "summary": s.get("summary", "")[:120]}
|
||
for s in top_scored
|
||
]
|
||
|
||
# Top recent news headlines
|
||
top_news = [{"title": n.get("title", ""), "impact": n.get("impact_score", 0)}
|
||
for n in sorted(news, key=lambda x: -(x.get("impact_score") or 0))[:5]]
|
||
|
||
import json
|
||
|
||
def gv(key: str) -> str:
|
||
v = gauges.get(key, {}).get("value")
|
||
return str(round(v, 2)) if v is not None else "N/A"
|
||
|
||
def gc(key: str) -> str:
|
||
v = gauges.get(key, {}).get("change_pct")
|
||
return f"{v:+.2f}%" if v is not None else "N/A"
|
||
|
||
prompt = f"""Tu es un stratège macro-géopolitique senior qui analyse la performance de notre système de détection de patterns.
|
||
|
||
CONTEXTE DU CYCLE (maintenant):
|
||
- Régime dominant: {dominant.upper()} (score: {scenarios.get('scores', {}).get(dominant, 0)}%)
|
||
- Score risque géopolitique: {geo_score_val}/100
|
||
- VIX: {gv('vix')} | Pente 10Y-3M: {gv('slope_10y3m')}% | DXY: {gc('dxy')} | Brent: {gc('brent')}
|
||
- Cuivre: {gc('copper')} | Or: {gc('gold')} | S&P vs 200j: {gv('spx_vs_200d')}%
|
||
|
||
TOP 5 PATTERNS ACTUELLEMENT LES MIEUX SCORÉS:
|
||
{json.dumps(top_scored_summary, ensure_ascii=False, indent=2)}
|
||
|
||
TRADES LOGUÉS CES 7 DERNIERS JOURS:
|
||
{json.dumps(trade_summary, ensure_ascii=False, indent=2)}
|
||
|
||
NEWS GÉOPOLITIQUES À FORT IMPACT:
|
||
{json.dumps(top_news, ensure_ascii=False, indent=2)}
|
||
|
||
Écris un COMMENTAIRE DE CYCLE concis (4-6 phrases) pour un trader options:
|
||
1. Le régime macro confirme-t-il les patterns qui scorent le mieux ?
|
||
2. Y a-t-il des news ou événements qui expliquent un écart avec nos prévisions ?
|
||
3. Quels patterns/trades méritent attention (confirmation ou invalidation) ?
|
||
4. Une recommandation tactique pour le prochain cycle (3h)
|
||
|
||
Réponds UNIQUEMENT en JSON: {{"commentary": "<ton texte 4-6 phrases>", "key_risk": "<risque principal en 1 phrase>", "top_pattern": "<nom du pattern le plus pertinent maintenant>"}}"""
|
||
|
||
result = _chat(
|
||
"Tu es un stratège macro senior. Analyse concise et actionnable. JSON uniquement.",
|
||
prompt,
|
||
model="gpt-4o",
|
||
json_mode=True,
|
||
max_tokens=500,
|
||
)
|
||
if result and result.get("commentary"):
|
||
return json.dumps(result, ensure_ascii=False)
|
||
except Exception as e:
|
||
logger.warning(f"[Cycle] Commentary generation failed: {e}")
|
||
return None
|
||
|
||
|
||
# ── IV Gate ───────────────────────────────────────────────────────────────────
|
||
|
||
def _apply_iv_gate(
|
||
scored: List[Dict],
|
||
iv_snapshots: Dict[str, Dict],
|
||
) -> tuple:
|
||
"""
|
||
Filter ALERT-verdict trades from each scored pattern's suggested_trades/trade_rankings
|
||
before they reach log_trade_entries.
|
||
|
||
Returns (filtered_scored, blocked_trades_list).
|
||
IVR thresholds are read from app_config (iv_gate_ivr_high, iv_gate_ivr_extreme,
|
||
iv_gate_skew_threshold) so they can be tuned from the Config page.
|
||
"""
|
||
from services.options_technical_agent import assess_strategy_fit
|
||
|
||
# Read thresholds from config (fall back to hardcoded defaults)
|
||
try:
|
||
from services.database import get_config as _gc
|
||
_ivr_high = float(_gc("iv_gate_ivr_high") or 60)
|
||
_ivr_extreme = float(_gc("iv_gate_ivr_extreme") or 80)
|
||
_skew_thresh = float(_gc("iv_gate_skew_threshold") or 8)
|
||
except Exception:
|
||
_ivr_high, _ivr_extreme, _skew_thresh = 60.0, 80.0, 8.0
|
||
|
||
# Monkey-patch thresholds into the rule engine for this call
|
||
import services.options_technical_agent as _ota_mod
|
||
_orig_ivr_high = getattr(_ota_mod, "_IVR_HIGH", None)
|
||
try:
|
||
_ota_mod._IVR_HIGH = _ivr_high
|
||
_ota_mod._IVR_EXTREME = _ivr_extreme
|
||
_ota_mod._SKEW_THRESH = _skew_thresh
|
||
except Exception:
|
||
pass
|
||
|
||
# Detect weekend: IVR=100 on weekends is artificial (weekend premium) — ignore rank
|
||
from datetime import datetime as _dt_cls
|
||
_weekend_now = _dt_cls.utcnow().weekday() >= 5
|
||
|
||
all_blocked: List[Dict] = []
|
||
|
||
for sp in scored:
|
||
trade_key = "trade_rankings" if "trade_rankings" in sp else "suggested_trades"
|
||
trades = sp.get(trade_key) or []
|
||
allowed = []
|
||
|
||
for trade in trades:
|
||
ticker = (trade.get("underlying") or trade.get("ticker") or "").upper()
|
||
strategy = trade.get("strategy") or "Long Call"
|
||
snap = iv_snapshots.get(ticker, {})
|
||
|
||
if not snap:
|
||
# No IV data → can't block; allow but don't judge
|
||
allowed.append(trade)
|
||
continue
|
||
|
||
iv_rank = snap.get("iv_rank")
|
||
# On weekends, ATM IV is inflated by weekend premium (market makers can't hedge).
|
||
# IVR=99-100 on a weekend is almost always artificial — don't block on it.
|
||
if _weekend_now and iv_rank is not None and iv_rank >= 99.0:
|
||
iv_rank = None
|
||
snap = {**snap, "iv_rank": None}
|
||
iv_current_pct = snap.get("iv_current_pct")
|
||
iv_min_52w = snap.get("iv_min_52w_pct")
|
||
iv_max_52w = snap.get("iv_max_52w_pct")
|
||
skew_pct = (snap.get("skew") or {}).get("skew_pct")
|
||
term_structure = (snap.get("term_structure") or {}).get("structure")
|
||
flow_bias = (snap.get("options_flow") or {}).get("flow_bias")
|
||
|
||
result = assess_strategy_fit(
|
||
strategy=strategy,
|
||
iv_rank=iv_rank,
|
||
iv_current_pct=iv_current_pct,
|
||
iv_min_52w=iv_min_52w,
|
||
iv_max_52w=iv_max_52w,
|
||
skew_pct=skew_pct,
|
||
term_structure=term_structure,
|
||
flow_bias=flow_bias,
|
||
)
|
||
|
||
if result["verdict"] == "ALERT":
|
||
blocked_entry = {
|
||
"ticker": ticker,
|
||
"strategy": strategy,
|
||
"pattern_id": sp.get("pattern_id"),
|
||
"pattern_name": sp.get("geo_trigger") or sp.get("pattern_name") or "",
|
||
"score": sp.get("score", 0),
|
||
"asset_class": trade.get("asset_class") or sp.get("asset_class") or "",
|
||
"iv_rank": iv_rank,
|
||
"iv_issues": result["issues"],
|
||
"optimal_strategy": result["optimal_strategy"],
|
||
"fit_score": result["fit_score"],
|
||
}
|
||
all_blocked.append(blocked_entry)
|
||
logger.warning(
|
||
f"[IVGate] BLOCKED {ticker} {strategy} (IVR={iv_rank}, score={result['fit_score']}): "
|
||
f"{result['issues'][0][:80] if result['issues'] else 'ALERT'}"
|
||
)
|
||
else:
|
||
allowed.append(trade)
|
||
|
||
sp[trade_key] = allowed
|
||
|
||
return scored, all_blocked
|
||
|
||
|
||
# ── Cycle report generation ───────────────────────────────────────────────────
|
||
|
||
def _generate_cycle_report(
|
||
run_id: str,
|
||
scored: List[Dict],
|
||
dominant: str,
|
||
scenarios: Dict,
|
||
geo_score_val: int,
|
||
news: List[Dict],
|
||
gauges: Dict,
|
||
ai_key: str,
|
||
added_patterns: List[Dict],
|
||
scoring_run_id: str,
|
||
portfolio_monitor: Optional[Dict],
|
||
commentary: Optional[str],
|
||
options_assessment: Optional[Dict] = None,
|
||
wavelet_signals: Optional[List[Dict]] = None,
|
||
) -> Optional[Dict]:
|
||
"""
|
||
Build the full cycle report dict:
|
||
- PnL + VaR snapshots (timestamped with this cycle)
|
||
- Delta vs previous cycle (patterns added, trades logged, trades closed)
|
||
- GPT-4o context narrative (what was in context, what it used, why)
|
||
- Risk summary
|
||
"""
|
||
import json as _json
|
||
|
||
try:
|
||
from services.database import get_cycle_runs, get_trade_entries_by_run
|
||
except ImportError:
|
||
from services.database import get_cycle_runs
|
||
get_trade_entries_by_run = None
|
||
|
||
# ── Get previous cycle timestamp ──────────────────────────────────────────
|
||
prev_cycle_at: Optional[str] = None
|
||
try:
|
||
history = get_cycle_runs(limit=5)
|
||
prev_cycles = [r for r in history if r["run_id"] != run_id and r.get("status") == "completed"]
|
||
if prev_cycles:
|
||
prev_cycle_at = prev_cycles[0].get("started_at")
|
||
except Exception:
|
||
pass
|
||
|
||
# ── PnL snapshot ──────────────────────────────────────────────────────────
|
||
pnl_snapshot_id: Optional[int] = None
|
||
pnl_summary: Dict = {}
|
||
try:
|
||
from services.var_service import save_pnl_snapshot, get_latest_pnl_snapshot
|
||
pnl_snapshot_id = save_pnl_snapshot()
|
||
snap = get_latest_pnl_snapshot()
|
||
if snap:
|
||
pnl_summary = {
|
||
"total_pnl_pct": snap.get("total_pnl_pct"),
|
||
"total_pnl_eur": snap.get("total_pnl_eur"),
|
||
"total_capital_eur": snap.get("total_capital_eur"),
|
||
"n_open": snap.get("n_open"),
|
||
"n_closed": snap.get("n_closed"),
|
||
}
|
||
logger.info(f"[CycleReport] PnL snapshot saved id={pnl_snapshot_id}")
|
||
except Exception as _e:
|
||
logger.warning(f"[CycleReport] PnL snapshot failed: {_e}")
|
||
|
||
# ── VaR snapshot ──────────────────────────────────────────────────────────
|
||
var_snapshot_id: Optional[int] = None
|
||
var_summary: Dict = {}
|
||
try:
|
||
from services.var_service import compute_var, save_var_snapshot
|
||
_var_cfg = get_cycle_step_config()["var_snapshot"]
|
||
var_result = compute_var(
|
||
confidence=_var_cfg["confidence"], horizon_days=_var_cfg["horizon_days"],
|
||
lookback_days=_var_cfg["lookback_days"], default_iv=_var_cfg["default_iv"],
|
||
)
|
||
if "error" not in var_result:
|
||
var_snapshot_id = save_var_snapshot(var_result, 0.95, 1, 252, 0.20)
|
||
var_summary = {
|
||
"hist_var_1d_pct": var_result.get("hist_var_1d_pct"),
|
||
"hist_cvar_pct": var_result.get("hist_cvar_pct"),
|
||
"mc_var_1d_pct": var_result.get("mc_var_1d_pct"),
|
||
"n_positions": var_result.get("n_positions"),
|
||
}
|
||
logger.info(f"[CycleReport] VaR snapshot saved id={var_snapshot_id}")
|
||
except Exception as _e:
|
||
logger.warning(f"[CycleReport] VaR snapshot failed: {_e}")
|
||
|
||
# ── Trades delta ──────────────────────────────────────────────────────────
|
||
trades_logged: List[Dict] = []
|
||
trades_closed: List[Dict] = []
|
||
try:
|
||
from services.database import get_conn
|
||
_conn = get_conn()
|
||
# Trades logged this cycle (by scoring_run_id)
|
||
_rows = _conn.execute(
|
||
"""SELECT underlying, strategy, entry_date, pnl_pct, score_at_entry, pattern_name
|
||
FROM trade_entry_prices WHERE run_id=? ORDER BY entry_date DESC""",
|
||
(scoring_run_id,),
|
||
).fetchall()
|
||
trades_logged = [dict(r) for r in _rows]
|
||
|
||
# Trades closed since previous cycle
|
||
if prev_cycle_at:
|
||
_closed = _conn.execute(
|
||
"""SELECT underlying, strategy, closed_at, pnl_pct, pattern_name
|
||
FROM trade_entry_prices
|
||
WHERE status='closed' AND closed_at >= ?
|
||
ORDER BY closed_at DESC""",
|
||
(prev_cycle_at,),
|
||
).fetchall()
|
||
trades_closed = [dict(r) for r in _closed]
|
||
_conn.close()
|
||
except Exception as _e:
|
||
logger.warning(f"[CycleReport] Delta trades query failed: {_e}")
|
||
|
||
# ── Context narrative (GPT-4o) ────────────────────────────────────────────
|
||
context_narrative: Dict = {}
|
||
try:
|
||
from services.ai_analyzer import _chat
|
||
|
||
top_news_ctx = [
|
||
{"title": n.get("title", "")[:100], "impact": round(float(n.get("impact_score") or 0), 2)}
|
||
for n in sorted(news, key=lambda x: -(float(x.get("impact_score") or 0)))[:8]
|
||
]
|
||
|
||
def _gv(k: str) -> str:
|
||
v = gauges.get(k, {}).get("value")
|
||
return str(round(v, 2)) if v is not None else "N/A"
|
||
|
||
def _gc(k: str) -> str:
|
||
v = gauges.get(k, {}).get("change_pct")
|
||
return f"{v:+.2f}%" if v is not None else "N/A"
|
||
|
||
new_pattern_names = [p.get("name", "") for p in added_patterns]
|
||
top_scored = sorted(scored, key=lambda x: -(x.get("score") or 0))[:5]
|
||
|
||
ctx_prompt = f"""Tu es un analyste stratégique qui documente le RAISONNEMENT d'un cycle d'analyse géo-macro.
|
||
|
||
CONTEXTE TRANSMIS À L'IA CE CYCLE:
|
||
- Régime macro dominant: {dominant.upper()} (score: {scenarios.get('scores', {}).get(dominant, 0)}%)
|
||
- Score risque géopolitique: {geo_score_val}/100
|
||
- VIX: {_gv('vix')} | Pente 10Y-3M: {_gv('slope_10y3m')} | S&P/200j: {_gv('spx_vs_200d')}%
|
||
- Cuivre: {_gc('copper')} | Or: {_gc('gold')} | DXY: {_gc('dxy')} | Brent: {_gc('brent')}
|
||
- Top 8 news géopolitiques transmises: {_json.dumps(top_news_ctx, ensure_ascii=False)}
|
||
|
||
NOUVELLES IDÉES GÉNÉRÉES CE CYCLE ({len(new_pattern_names)} patterns ajoutés):
|
||
{_json.dumps(new_pattern_names, ensure_ascii=False)}
|
||
|
||
TOP 5 PATTERNS LES MIEUX SCORÉS MAINTENANT:
|
||
{_json.dumps([{"name": s.get("geo_trigger","?"), "score": s.get("score"), "catalyst": s.get("key_catalyst","")[:80]} for s in top_scored], ensure_ascii=False)}
|
||
|
||
Ta tâche: Explique le RAISONNEMENT de ce cycle.
|
||
Pour chaque nouvelle idée générée:
|
||
1. Quels éléments SPÉCIFIQUES du contexte transmis (news, macro, gauges) l'ont motivée ?
|
||
2. Qu'est-ce qui vient de ta connaissance GÉNÉRALE des marchés (pas du contexte transmis) ?
|
||
3. Quels signaux étaient les plus déterminants ?
|
||
|
||
Réponds en JSON avec ce schéma EXACT:
|
||
{{
|
||
"narrative": "<texte narratif 6-10 phrases expliquant le raisonnement global du cycle>",
|
||
"context_driven": ["<signal du contexte qui a motivé une idée spécifique>", ...],
|
||
"general_knowledge": ["<connaissance générale utilisée>", ...],
|
||
"key_signals": ["<3-5 signaux les plus décisifs>"],
|
||
"context_log": {{
|
||
"macro_regime": "{dominant}",
|
||
"geo_score": {geo_score_val},
|
||
"dominant_news": ["<titre news 1>", "<titre news 2>", "<titre news 3>"],
|
||
"key_gauges": {{"vix": {_gv('vix')}, "slope": {_gv('slope_10y3m')}, "copper_chg": {_gc('copper')}, "gold_chg": {_gc('gold')}}}
|
||
}}
|
||
}}"""
|
||
|
||
result = _chat(
|
||
"Tu es un analyste macro-géopolitique. Documente le raisonnement du cycle en JSON.",
|
||
ctx_prompt,
|
||
model="gpt-4o",
|
||
json_mode=True,
|
||
max_tokens=800,
|
||
)
|
||
if result and result.get("narrative"):
|
||
context_narrative = result
|
||
logger.info(f"[CycleReport] Context narrative generated ({len(result.get('narrative',''))} chars)")
|
||
except Exception as _e:
|
||
logger.warning(f"[CycleReport] Context narrative failed: {_e}")
|
||
context_narrative = {"narrative": "", "context_driven": [], "general_knowledge": [], "key_signals": []}
|
||
|
||
# ── Parse existing commentary ─────────────────────────────────────────────
|
||
commentary_parsed: Dict = {}
|
||
if commentary:
|
||
try:
|
||
commentary_parsed = _json.loads(commentary) if isinstance(commentary, str) else commentary
|
||
except Exception:
|
||
commentary_parsed = {"commentary": str(commentary)}
|
||
|
||
# ── Calibration report ───────────────────────────────────────────────────
|
||
calibration_report: Dict = {}
|
||
try:
|
||
from services.database import get_calibration_summary as _get_calib
|
||
calib_rows = _get_calib()
|
||
if calib_rows:
|
||
pure_ai = [r for r in calib_rows if r["source"] == "pure_ai"]
|
||
early = [r for r in calib_rows if r["source"] == "early"]
|
||
mixed = [r for r in calib_rows if r["source"] == "mixed"]
|
||
driven = [r for r in calib_rows if r["source"] == "data_driven"]
|
||
with_data = [r for r in calib_rows if (r["n_mature_trades"] or 0) > 0]
|
||
avg_w = (
|
||
round(sum(r["calibration_weight"] for r in with_data) / len(with_data), 3)
|
||
if with_data else 0.0
|
||
)
|
||
calibration_report = {
|
||
"total_patterns": len(calib_rows),
|
||
"pure_ai_count": len(pure_ai),
|
||
"early_count": len(early),
|
||
"mixed_count": len(mixed),
|
||
"data_driven_count": len(driven),
|
||
"avg_calibration_weight": avg_w,
|
||
"avg_calibration_weight_pct": round(avg_w * 100, 1),
|
||
"patterns_with_data": len(with_data),
|
||
"detail": [
|
||
{
|
||
"name": r["pattern_name"],
|
||
"asset_class": r["asset_class"],
|
||
"source": r["source"],
|
||
"weight_pct": r["calibration_weight_pct"],
|
||
"ai_estimate": r["ai_estimate"],
|
||
"observed": r["observed_avg_win_pct"],
|
||
"calibrated": r["calibrated_expected_move"],
|
||
"n_trades": r["n_mature_trades"],
|
||
"win_rate_pct": round((r["bayes_win_rate"] or 0) * 100, 1),
|
||
}
|
||
for r in calib_rows if (r["n_mature_trades"] or 0) > 0
|
||
][:15],
|
||
}
|
||
logger.info(
|
||
f"[CycleReport] Calibration: {len(pure_ai)} pure-AI, "
|
||
f"{len(mixed)+len(driven)} with data, avg_weight={avg_w:.1%}"
|
||
)
|
||
except Exception as _ce:
|
||
logger.warning(f"[CycleReport] Calibration report failed: {_ce}")
|
||
|
||
# ── Assemble full report ──────────────────────────────────────────────────
|
||
report = {
|
||
"run_id": run_id,
|
||
"macro_dominant": dominant,
|
||
"geo_score": geo_score_val,
|
||
"macro_scores": scenarios.get("scores", {}),
|
||
"macro_reasons": scenarios.get("reasons", {}).get(dominant, []),
|
||
# Snapshots
|
||
"pnl_snapshot_id": pnl_snapshot_id,
|
||
"var_snapshot_id": var_snapshot_id,
|
||
"pnl_summary": pnl_summary,
|
||
"var_summary": var_summary,
|
||
# Delta
|
||
"patterns_added": len(added_patterns),
|
||
"patterns_added_list": [
|
||
{"name": p.get("name"), "description": (p.get("description") or "")[:150],
|
||
"asset_class": p.get("asset_class"), "macro_fit": p.get("macro_fit"),
|
||
"triggers": p.get("triggers", [])[:3]}
|
||
for p in added_patterns
|
||
],
|
||
"trades_logged": len(trades_logged),
|
||
"trades_logged_list": trades_logged[:10],
|
||
"trades_closed": len(trades_closed),
|
||
"trades_closed_list": trades_closed[:10],
|
||
"prev_cycle_at": prev_cycle_at,
|
||
# Context narrative
|
||
"context_narrative": context_narrative,
|
||
# Cycle commentary (existing)
|
||
"commentary": commentary_parsed,
|
||
# Risk
|
||
"portfolio_monitor": portfolio_monitor,
|
||
"risk_alerts": portfolio_monitor.get("alerts") if portfolio_monitor else None,
|
||
# Wavelet signals detected this cycle (watchlist scan, see services.wavelet_signals)
|
||
"wavelet_signals": wavelet_signals or [],
|
||
# Top scored patterns
|
||
"top_scored": [
|
||
{"name": s.get("geo_trigger"), "score": s.get("score"),
|
||
"summary": (s.get("summary") or "")[:120], "key_catalyst": s.get("key_catalyst", "")}
|
||
for s in sorted(scored, key=lambda x: -(x.get("score") or 0))[:5]
|
||
],
|
||
# Options Technical Agent assessment
|
||
"options_technical": {
|
||
"global_assessment": (options_assessment or {}).get("global_assessment", ""),
|
||
"global_score": (options_assessment or {}).get("global_score"),
|
||
"n_ok": (options_assessment or {}).get("n_ok", 0),
|
||
"n_warn": (options_assessment or {}).get("n_warn", 0),
|
||
"n_alert": (options_assessment or {}).get("n_alert", 0),
|
||
"assessments": (options_assessment or {}).get("assessments", []),
|
||
} if options_assessment is not None else None,
|
||
# Calibration: AI vs observed expected_move blend
|
||
"calibration_report": calibration_report,
|
||
}
|
||
return report
|
||
|
||
|
||
def generate_standalone_report() -> Dict[str, Any]:
|
||
"""Cycle Actions — standalone "generate-report" action. _generate_cycle_report()
|
||
itself is NOT modified (too tightly coupled to run_cycle_once()'s in-memory
|
||
state to safely change) — instead every one of its parameters is
|
||
reconstructed from an independent, DB-or-live source:
|
||
- scored <- get_last_scores() (config key "last_pattern_scores",
|
||
filled by save_pattern_scores() every real cycle)
|
||
- news/geo_score <- same live sequence as cycle Step 1
|
||
- dominant/scenarios/gauges <- same sequence as the update-regime action
|
||
- commentary <- _generate_cycle_commentary(), independently callable
|
||
- wavelet_signals <- compute_and_save_wavelet_signals(), independently callable
|
||
- portfolio_monitor <- analyze_simulation_portfolio() (+ AI only if alerts)
|
||
- added_patterns/options_assessment <- cycle-only artifacts, not
|
||
reconstructible after the fact -> empty/None (report is honest, just
|
||
thinner on these two fields than a live cycle's report)."""
|
||
from services.database import get_config, get_pattern_scores, save_cycle_report
|
||
from services.data_fetcher import fetch_geo_news, get_macro_gauges, score_macro_scenarios
|
||
from services.geo_analyzer import compute_geo_risk_score
|
||
from services.ai_analyzer import ai_score_news_batch, ai_score_geo_risk
|
||
|
||
run_id = datetime.utcnow().isoformat()
|
||
ai_key = get_config("openai_api_key") or ""
|
||
|
||
last_scores = get_pattern_scores()
|
||
scored = last_scores.get("scores") or []
|
||
scoring_run_id = last_scores.get("run_id") or run_id
|
||
|
||
news = ai_score_news_batch(fetch_geo_news())
|
||
algo_geo = compute_geo_risk_score(news)
|
||
try:
|
||
geo_obj = ai_score_geo_risk(news, algo_geo, log_meta={"run_id": run_id, "call_type": "geo_risk_score"})
|
||
except Exception:
|
||
geo_obj = algo_geo
|
||
geo_score_val = int(round(geo_obj.get("score") or 0))
|
||
|
||
gauges = get_macro_gauges()
|
||
scenarios = score_macro_scenarios(gauges)
|
||
dominant = scenarios.get("dominant", "incertain")
|
||
|
||
commentary = None
|
||
try:
|
||
commentary = _generate_cycle_commentary(scored, dominant, scenarios, geo_score_val, news, gauges)
|
||
except Exception as e:
|
||
logger.warning(f"[StandaloneReport] Commentary generation failed: {e}")
|
||
|
||
wavelet_signals: List[Dict] = []
|
||
try:
|
||
from services.wavelet_signals import compute_and_save_wavelet_signals
|
||
wavelet_signals = compute_and_save_wavelet_signals(run_id)
|
||
except Exception as e:
|
||
logger.warning(f"[StandaloneReport] Wavelet scan failed (non-blocking): {e}")
|
||
|
||
portfolio_monitor = None
|
||
try:
|
||
from services.portfolio_risk import analyze_simulation_portfolio
|
||
risk = analyze_simulation_portfolio()
|
||
if risk.get("alerts"):
|
||
portfolio_monitor = _run_portfolio_monitor(risk, run_id)
|
||
except Exception as e:
|
||
logger.warning(f"[StandaloneReport] Portfolio monitor failed (non-blocking): {e}")
|
||
|
||
report = _generate_cycle_report(
|
||
run_id=run_id, scored=scored, dominant=dominant, scenarios=scenarios,
|
||
geo_score_val=geo_score_val, news=news, gauges=gauges, ai_key=ai_key,
|
||
added_patterns=[], scoring_run_id=scoring_run_id,
|
||
portfolio_monitor=portfolio_monitor, commentary=commentary,
|
||
options_assessment=None, wavelet_signals=wavelet_signals,
|
||
)
|
||
if report:
|
||
save_cycle_report(run_id, report)
|
||
return {"run_id": run_id, "report": report}
|
||
|
||
|
||
# ── Auto portfolio snapshot ───────────────────────────────────────────────────
|
||
|
||
def _auto_portfolio_snapshot(ai_key: str) -> None:
|
||
"""
|
||
Called in a background thread at the end of each cycle.
|
||
Fetches live prices, checks if enough trades are priced (P&L ≠ 0),
|
||
and if so generates a GPT-4o portfolio report so the NEXT cycle has
|
||
fresh performance lessons. Skipped silently if not enough data.
|
||
"""
|
||
try:
|
||
import os
|
||
os.environ["OPENAI_API_KEY"] = ai_key
|
||
|
||
from services.database import (
|
||
get_mtm_trades_with_traces, save_ai_report, _trade_maturity,
|
||
)
|
||
from datetime import date as _date
|
||
|
||
data = get_mtm_trades_with_traces(days=get_cycle_step_config()["iv_context"]["lookback_days"], limit_movers=10)
|
||
all_trades = data.get("all_trades", [])
|
||
priced = data.get("priced_count", 0)
|
||
|
||
# Classify each trade by maturity
|
||
def _with_maturity(t: Dict) -> Dict:
|
||
try:
|
||
dh = (_date.today() - _date.fromisoformat(t["entry_date"])).days
|
||
except Exception:
|
||
dh = 0
|
||
mat = _trade_maturity(dh, t.get("horizon_days") or 90)
|
||
return {**t, "days_held": dh, "maturity": mat}
|
||
|
||
enriched = [_with_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")]
|
||
|
||
# Only learn lessons from mature trades (weight > 0.9)
|
||
meaningful_mature = [t for t in matures if abs(t.get("pnl_pct", 0)) > 0.1]
|
||
|
||
if len(meaningful_mature) < 2:
|
||
logger.info(
|
||
f"[AutoSnapshot] Skipping GPT-4o report: only {len(meaningful_mature)} mature trades "
|
||
f"with meaningful P&L (need ≥ 2). "
|
||
f"Immatures={len(trop_tot)}, en cours={len(en_cours)}, matures={len(matures)}"
|
||
)
|
||
# Still try to refresh Super Contexte (independent of portfolio report)
|
||
_auto_synthesize_knowledge(ai_key)
|
||
return
|
||
|
||
# Sort matures by P&L for the 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
|
||
|
||
stats = {
|
||
"total_trades": len(all_trades),
|
||
"priced_count": priced,
|
||
"avg_pnl_pct": avg_pnl,
|
||
"mature_count": len(matures),
|
||
"early_count": len(trop_tot) + len(en_cours),
|
||
}
|
||
|
||
# Build maturity context blocks for the prompt
|
||
def _trade_line(t: Dict) -> str:
|
||
mat = t["maturity"]
|
||
return (
|
||
f" {t.get('underlying','?')} {t.get('strategy','?')} "
|
||
f"P&L={t.get('pnl_pct',0):+.1f}% "
|
||
f"[{mat['readable']}] "
|
||
f"score_entrée={t.get('score_at_entry','?')}"
|
||
)
|
||
|
||
mature_wins_block = "\n".join(_trade_line(t) for t in winners) or " Aucun"
|
||
mature_losses_block = "\n".join(_trade_line(t) for t in losers) or " Aucun"
|
||
early_block = "\n".join(
|
||
f" {t.get('underlying','?')} {t.get('strategy','?')} "
|
||
f"P&L={t.get('pnl_pct',0):+.1f}% [{t['maturity']['readable']}]"
|
||
for t in (trop_tot + en_cours)[:6]
|
||
) or " Aucun"
|
||
|
||
avg_str = f"{avg_pnl:+.1f}%" if avg_pnl is not None else "N/A"
|
||
from services.ai_analyzer import _chat
|
||
|
||
prompt = f"""Tu es un stratège macro-géopolitique senior. Rapport post-cycle automatique.
|
||
|
||
⚠️ RÈGLE FONDAMENTALE DE TIMING :
|
||
Les options ont des horizons de 30-90 jours. Un trade de 3 jours ne dit RIEN sur sa performance finale.
|
||
Tu dois UNIQUEMENT tirer des leçons des trades MATURES (≥35% de l'horizon écoulé).
|
||
Les trades immatures sont listés pour information seulement — n'en tire AUCUNE conclusion de performance.
|
||
|
||
═══ STATISTIQUES GLOBALES ═══
|
||
Période analysée : 90 jours | Trades total: {len(all_trades)} | Pricés: {priced}
|
||
Trades MATURES (signal fiable): {len(matures)} | P&L moyen matures: {avg_str}
|
||
Trades IMMATURES (trop tôt pour juger): {len(trop_tot) + len(en_cours)}
|
||
|
||
═══ MATURES — TOP GAINS (signal fiable) ═══
|
||
{mature_wins_block}
|
||
|
||
═══ MATURES — TOP PERTES (signal fiable) ═══
|
||
{mature_losses_block}
|
||
|
||
═══ EN COURS — NE PAS ÉVALUER (trop tôt) ═══
|
||
{early_block}
|
||
|
||
Génère un rapport JSON basé UNIQUEMENT sur les trades matures :
|
||
{{
|
||
"headline": "<1 phrase résumant la performance des trades matures>",
|
||
"regime_assessment": "<alignement régime macro avec nos thèses matures ?>",
|
||
"winners_analysis": "<pourquoi les trades matures gagnants ont marché — 2-3 phrases>",
|
||
"losers_analysis": "<pourquoi les trades matures perdants ont déçu — 2-3 phrases>",
|
||
"key_lessons": ["<leçon 1 tirée 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 pour le prochain cycle>",
|
||
"risk_watch": "<1-2 risques à surveiller>",
|
||
"timing_note": "<observation sur les trades immatures — à surveiller mais pas à juger>"
|
||
}}"""
|
||
|
||
result = _chat(
|
||
"Tu es un stratège macro senior. Rapport post-cycle concis. JSON uniquement.",
|
||
prompt,
|
||
model="gpt-4o",
|
||
json_mode=True,
|
||
max_tokens=800,
|
||
)
|
||
if not result:
|
||
logger.warning("[AutoSnapshot] GPT-4o returned empty response")
|
||
return
|
||
|
||
report_id = save_ai_report(
|
||
days=30, stats=stats, winners=winners, losers=losers, report=result,
|
||
report_type="portfolio",
|
||
)
|
||
logger.info(
|
||
f"[AutoSnapshot] Portfolio report #{report_id} saved automatically "
|
||
f"({len(meaningful_mature)} meaningful mature trades, avg P&L {avg_str})"
|
||
)
|
||
|
||
# ── Auto-synthesize Super Contexte if stale (>6h or never generated) ──
|
||
_auto_synthesize_knowledge(ai_key)
|
||
|
||
except Exception as e:
|
||
logger.error(f"[AutoSnapshot] Failed: {e}", exc_info=True)
|
||
|
||
|
||
def _auto_synthesize_knowledge(ai_key: str) -> None:
|
||
"""
|
||
Synthesize the Super Contexte knowledge base after a portfolio report is generated.
|
||
Skipped if the last synthesis is < 6 hours old, to avoid redundant GPT-4o calls.
|
||
"""
|
||
try:
|
||
import os, json as _json
|
||
from datetime import datetime as _dt, timedelta as _td
|
||
import openai
|
||
|
||
os.environ["OPENAI_API_KEY"] = ai_key
|
||
|
||
from services.database import (
|
||
get_latest_reasoning_state, save_reasoning_state,
|
||
get_all_kb_entries, save_kb_entry,
|
||
list_ai_reports, get_mtm_trades_with_traces,
|
||
)
|
||
|
||
synth_cfg = get_cycle_step_config()["knowledge_synthesis"]
|
||
if not synth_cfg["enabled"]:
|
||
logger.info("[AutoSynth] Disabled via cycle_step_config — skipping")
|
||
return
|
||
|
||
# Skip if last synthesis is younger than the configured threshold
|
||
min_hours = synth_cfg["min_hours_between"]
|
||
last_state = get_latest_reasoning_state()
|
||
if last_state:
|
||
try:
|
||
last_at = _dt.fromisoformat(last_state["created_at"])
|
||
age_h = (_dt.utcnow() - last_at).total_seconds() / 3600
|
||
if age_h < min_hours:
|
||
logger.info(
|
||
f"[AutoSynth] Super Contexte is {age_h:.1f}h old — skipping re-synthesis (threshold: {min_hours}h)"
|
||
)
|
||
return
|
||
except Exception:
|
||
pass
|
||
|
||
reports = list_ai_reports(limit=10)
|
||
mtm_data = get_mtm_trades_with_traces(days=get_cycle_step_config()["iv_context"]["lookback_days"])
|
||
trades = mtm_data.get("all_trades", []) if isinstance(mtm_data, dict) else []
|
||
kb_entries = get_all_kb_entries()
|
||
|
||
logger.info(
|
||
f"[AutoSynth] Starting Super Contexte synthesis "
|
||
f"({len(reports)} rapports, {len(trades)} trades, {len(kb_entries)} KB entries)"
|
||
)
|
||
|
||
# Build synthesis prompt (reuse router logic inline to avoid import cycle)
|
||
from routers.knowledge import _build_synthesis_prompt
|
||
system_msg, user_msg = _build_synthesis_prompt(reports, trades, kb_entries)
|
||
|
||
client = openai.OpenAI(api_key=ai_key)
|
||
resp = client.chat.completions.create(
|
||
model="gpt-4o",
|
||
messages=[
|
||
{"role": "system", "content": system_msg},
|
||
{"role": "user", "content": user_msg},
|
||
],
|
||
temperature=0.3,
|
||
max_tokens=2500,
|
||
response_format={"type": "json_object"},
|
||
)
|
||
raw = resp.choices[0].message.content or "{}"
|
||
synthesis = _json.loads(raw)
|
||
narrative = synthesis.pop("narrative", "Synthèse non disponible.")
|
||
|
||
state_id = save_reasoning_state(
|
||
narrative=narrative,
|
||
synthesis=synthesis,
|
||
sources_count=len(reports) + len(trades),
|
||
reports_used=len(reports),
|
||
trades_analyzed=len(trades),
|
||
)
|
||
|
||
# Auto-persist new KB entries from synthesis
|
||
added = 0
|
||
for regime in synthesis.get("regime_insights", []):
|
||
if regime.get("observation"):
|
||
save_kb_entry("régimes", f"Régime: {regime.get('regime','?')}", regime["observation"],
|
||
regime.get("confidence", 50), "auto-synth")
|
||
added += 1
|
||
for pattern in synthesis.get("pattern_insights", []):
|
||
if pattern.get("observation"):
|
||
save_kb_entry("patterns", f"Pattern: {pattern.get('pattern','?')}", pattern["observation"],
|
||
pattern.get("confidence", 50), "auto-synth")
|
||
added += 1
|
||
for mistake in synthesis.get("recurring_mistakes", []):
|
||
if mistake.get("mistake"):
|
||
save_kb_entry("erreurs", mistake["mistake"][:80],
|
||
f"{mistake.get('mistake','')} → {mistake.get('mitigation','')}",
|
||
70, "auto-synth")
|
||
added += 1
|
||
|
||
logger.info(
|
||
f"[AutoSynth] Super Contexte v{state_id} saved automatically "
|
||
f"({added} KB entries added)"
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"[AutoSynth] Failed: {e}", exc_info=True)
|
||
|
||
|
||
# ── Scheduler ─────────────────────────────────────────────────────────────────
|
||
|
||
def _parse_weekend_times(times_str: str):
|
||
"""Parse 'HH:MM,HH:MM' into list of (hour, minute) tuples."""
|
||
result = []
|
||
for t in times_str.split(","):
|
||
t = t.strip()
|
||
try:
|
||
parts = t.split(":")
|
||
result.append((int(parts[0]), int(parts[1])))
|
||
except Exception:
|
||
pass
|
||
return result or [(8, 0), (22, 0)]
|
||
|
||
|
||
def _next_weekend_slot(now_utc: datetime, weekend_times):
|
||
"""Return (next_dt, wait_secs) for the next weekend cycle time (UTC)."""
|
||
from datetime import timedelta
|
||
today = now_utc.date()
|
||
candidates = []
|
||
for day_offset in range(3):
|
||
for h, m in weekend_times:
|
||
candidate = datetime(today.year, today.month, today.day, h, m) + timedelta(days=day_offset)
|
||
if candidate > now_utc + timedelta(minutes=5):
|
||
candidates.append(candidate)
|
||
if not candidates:
|
||
fallback = now_utc + timedelta(hours=12)
|
||
return fallback, 12 * 3600
|
||
next_slot = min(candidates)
|
||
return next_slot, max(60, (next_slot - now_utc).total_seconds())
|
||
|
||
|
||
def _scheduler_loop(stop_event: threading.Event):
|
||
"""Background loop that runs the cycle at the configured interval.
|
||
|
||
Weekday: fires every auto_cycle_hours.
|
||
Weekend: fires only at weekend_cycle_times (UTC HH:MM list).
|
||
If weekend_cycle_enabled=false, sleeps until Monday.
|
||
"""
|
||
from services.database import get_config
|
||
from datetime import timedelta
|
||
|
||
while not stop_event.is_set():
|
||
try:
|
||
interval_hours = float(get_config("auto_cycle_hours") or "3")
|
||
weekend_enabled = (get_config("weekend_cycle_enabled") or "true").lower() == "true"
|
||
weekend_times_str = get_config("weekend_cycle_times") or "08:00,22:00"
|
||
except Exception:
|
||
interval_hours, weekend_enabled, weekend_times_str = 3.0, True, "08:00,22:00"
|
||
|
||
weekend_times = _parse_weekend_times(weekend_times_str)
|
||
now_utc = datetime.utcnow()
|
||
weekday = now_utc.weekday()
|
||
is_weekend = weekday >= 5
|
||
|
||
if is_weekend and not weekend_enabled:
|
||
# Sleep until Monday 00:05 UTC
|
||
days_to_monday = 7 - weekday # Sat→2, Sun→1
|
||
next_monday = datetime(now_utc.year, now_utc.month, now_utc.day, 0, 5) + timedelta(days=days_to_monday)
|
||
wait_secs = max(60, (next_monday - now_utc).total_seconds())
|
||
_current_status["next_run_at"] = next_monday.isoformat()
|
||
_current_status["interval_hours"] = interval_hours
|
||
logger.info(f"[Scheduler] Weekend — cycles désactivés. Reprise lundi {next_monday.strftime('%Y-%m-%d %H:%M')} UTC")
|
||
stop_event.wait(timeout=wait_secs)
|
||
continue # don't run a cycle, loop back
|
||
|
||
if is_weekend:
|
||
next_slot, wait_secs = _next_weekend_slot(now_utc, weekend_times)
|
||
_current_status["next_run_at"] = next_slot.isoformat()
|
||
_current_status["interval_hours"] = interval_hours
|
||
logger.info(f"[Scheduler] Weekend — prochain cycle à {next_slot.strftime('%H:%M')} UTC ({wait_secs/3600:.1f}h)")
|
||
else:
|
||
# Account for time already elapsed since last cycle so a restart
|
||
# doesn't reset the countdown — fire as soon as the interval is due.
|
||
last_run_iso = _current_status.get("last_run_at")
|
||
if not last_run_iso:
|
||
# Also check DB for the most recent completed run
|
||
try:
|
||
from services.database import get_cycle_runs
|
||
recent = get_cycle_runs(limit=1)
|
||
if recent:
|
||
last_run_iso = recent[0].get("created_at") or recent[0].get("run_id", "")[:19].replace("T", " ")
|
||
except Exception:
|
||
pass
|
||
elapsed_secs = 0.0
|
||
if last_run_iso:
|
||
try:
|
||
last_dt = datetime.fromisoformat(last_run_iso.replace(" ", "T").rstrip("Z"))
|
||
elapsed_secs = (now_utc - last_dt).total_seconds()
|
||
except Exception:
|
||
pass
|
||
wait_secs = max(60, interval_hours * 3600 - elapsed_secs)
|
||
next_run = now_utc + timedelta(seconds=wait_secs)
|
||
_current_status["next_run_at"] = next_run.isoformat()
|
||
_current_status["interval_hours"] = interval_hours
|
||
logger.info(f"[Scheduler] Prochain cycle dans {wait_secs/3600:.1f}h ({next_run.strftime('%H:%M')} UTC) — elapsed since last: {elapsed_secs/3600:.1f}h")
|
||
|
||
stop_event.wait(timeout=wait_secs)
|
||
|
||
if stop_event.is_set():
|
||
break
|
||
|
||
# Check if still enabled
|
||
try:
|
||
enabled = (get_config("auto_cycle_enabled") or "false").lower() == "true"
|
||
except Exception:
|
||
enabled = False
|
||
|
||
if enabled:
|
||
logger.info("[Scheduler] Running scheduled auto-cycle")
|
||
try:
|
||
run_cycle_once(trigger="auto")
|
||
except Exception as e:
|
||
logger.error(f"[Scheduler] Cycle error: {e}", exc_info=True)
|
||
|
||
|
||
def start_scheduler():
|
||
"""Start the background scheduler thread if auto_cycle is enabled."""
|
||
global _cycle_thread, _stop_event
|
||
from services.database import get_config
|
||
|
||
enabled = (get_config("auto_cycle_enabled") or "false").lower() == "true"
|
||
_current_status["enabled"] = enabled
|
||
|
||
if not enabled:
|
||
logger.info("[Scheduler] Auto-cycle disabled — skipping scheduler start")
|
||
return
|
||
|
||
if _cycle_thread and _cycle_thread.is_alive():
|
||
logger.info("[Scheduler] Already running")
|
||
return
|
||
|
||
_stop_event = threading.Event()
|
||
_cycle_thread = threading.Thread(
|
||
target=_scheduler_loop,
|
||
args=(_stop_event,),
|
||
daemon=True,
|
||
name="auto-cycle-scheduler",
|
||
)
|
||
_cycle_thread.start()
|
||
logger.info("[Scheduler] Auto-cycle scheduler started")
|
||
|
||
|
||
def stop_scheduler():
|
||
"""Stop the background scheduler thread."""
|
||
global _cycle_thread
|
||
_stop_event.set()
|
||
if _cycle_thread:
|
||
_cycle_thread.join(timeout=5)
|
||
_current_status["enabled"] = False
|
||
logger.info("[Scheduler] Stopped")
|
||
|
||
|
||
def restart_scheduler():
|
||
"""Restart the scheduler — call after config changes."""
|
||
stop_scheduler()
|
||
_stop_event.clear()
|
||
start_scheduler()
|
||
|
||
|
||
def trigger_manual():
|
||
"""Run one cycle immediately in a background thread (non-blocking)."""
|
||
t = threading.Thread(target=run_cycle_once, args=("manual",), daemon=True, name="auto-cycle-manual")
|
||
t.start()
|
||
return t
|
||
|
||
|
||
def get_status() -> Dict[str, Any]:
|
||
from services.database import get_config, get_cycle_runs
|
||
try:
|
||
interval_hours = float(get_config("auto_cycle_hours") or "3")
|
||
enabled = (get_config("auto_cycle_enabled") or "false").lower() == "true"
|
||
sim_threshold = float(get_config("auto_cycle_similarity_threshold") or "0.30")
|
||
min_ev = float(get_config("min_ev_threshold") or "0.0")
|
||
min_score = int(get_config("min_score_threshold") or "0")
|
||
retention_days = int(get_config("journal_retention_days") or "90")
|
||
maturity_pct = int(get_config("maturity_threshold_pct") or "35")
|
||
trade_budget_eur = float(get_config("trade_budget_eur") or "5000")
|
||
preferred_horizon_min = int(get_config("preferred_horizon_min") or "30")
|
||
preferred_horizon_max = int(get_config("preferred_horizon_max") or "180")
|
||
except Exception:
|
||
interval_hours, enabled, sim_threshold, min_ev, min_score = 3.0, False, 0.30, 0.0, 0
|
||
retention_days, maturity_pct = 90, 35
|
||
trade_budget_eur, preferred_horizon_min, preferred_horizon_max = 5000.0, 30, 180
|
||
|
||
recent = get_cycle_runs(limit=1)
|
||
last = recent[0] if recent else None
|
||
|
||
# Enrich last_cycle with the scoring_run_id used for trade_entry_prices
|
||
if last:
|
||
try:
|
||
from services.database import get_conn
|
||
_conn = get_conn()
|
||
started = (last.get("started_at") or "").replace(" ", "T")
|
||
completed = last.get("completed_at") or ""
|
||
if started and completed:
|
||
row = _conn.execute(
|
||
"SELECT run_id FROM trade_entry_prices WHERE run_id >= ? AND run_id <= ?"
|
||
" ORDER BY run_id DESC LIMIT 1",
|
||
(started, completed),
|
||
).fetchone()
|
||
if row:
|
||
last = dict(last)
|
||
last["scoring_run_id"] = row["run_id"]
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
weekend_enabled = (get_config("weekend_cycle_enabled") or "true").lower() == "true"
|
||
weekend_cycle_times = get_config("weekend_cycle_times") or "08:00,22:00"
|
||
except Exception:
|
||
weekend_enabled, weekend_cycle_times = True, "08:00,22:00"
|
||
|
||
return {
|
||
**_current_status,
|
||
"enabled": enabled,
|
||
"interval_hours": interval_hours,
|
||
"similarity_threshold": sim_threshold,
|
||
"min_ev_threshold": min_ev,
|
||
"min_score_threshold": min_score,
|
||
"journal_retention_days": retention_days,
|
||
"maturity_threshold_pct": maturity_pct,
|
||
"trade_budget_eur": trade_budget_eur,
|
||
"preferred_horizon_min": preferred_horizon_min,
|
||
"preferred_horizon_max": preferred_horizon_max,
|
||
"weekend_cycle_enabled": weekend_enabled,
|
||
"weekend_cycle_times": weekend_cycle_times,
|
||
"cycle_step_config": get_cycle_step_config(),
|
||
"last_cycle": last,
|
||
"scheduler_alive": bool(_cycle_thread and _cycle_thread.is_alive()),
|
||
}
|