""" 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 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 # seuil cosinus pour considérer deux patterns comme doublons def _is_duplicate_pattern( candidate: Dict, existing: List[Dict], api_key: str, jaccard_threshold: float = 0.30, ) -> bool: """ Retourne True si le candidat est trop similaire à un pattern existant. Essaie les embeddings cosinus (Sprint 4.3) avec fallback Jaccard. """ 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 a fonctionné return sim >= _EMBED_SIM_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 # ── 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 = ( "Tu es un risk manager spécialisé dans les portefeuilles d'options géopolitiques. " "Analyse l'état du portefeuille simulé ci-dessous et génère des recommandations concrètes. " "Réponds en JSON avec ce schéma exact: " '{"assessment": "<2 phrases max sur l\'état global>", ' '"actions": [{"priority": "high|medium", "type": "close_trade|rebalance|monitor", ' '"trade_id": , "underlying": "", "reason": ""}], ' '"rebalance_suggestion": ""}' ) 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, ) 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, _chat, DEFAULT_ANALYSIS_TEMPLATE, ) # 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") add_cycle_run(run_id, trigger=trigger) _current_status["running"] = True _current_status["last_run_id"] = run_id # ── 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 geo_score_obj = compute_geo_risk_score(news) geo_score_val = int(geo_score_obj.get("score") or 0) summary["geo_score"] = geo_score_val # ── 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() 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 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}") try: from services.data_fetcher import get_economic_calendar calendar = get_economic_calendar() 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, ) 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): # 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.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: Collect IV context ───────────────────────────────────── iv_context = "" 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=90) _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)") template = get_config("analysis_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, ) 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", []) # ── 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) log_trade_entries(run_id=scoring_run_id, scored_patterns=scored, quotes=quotes) # ── Step 5.1: Portfolio monitor — conflict & concentration check ────── try: from services.portfolio_risk import analyze_simulation_portfolio _risk = analyze_simulation_portfolio() if _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 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) ────────────────────────── try: from services.database import detect_and_save_regime_clusters _cluster_result = detect_and_save_regime_clusters(n_clusters=4, days=180) 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 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", ) _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.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, ) 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 (last 7 days) entries = get_trade_entry_prices(7) 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": "", "key_risk": "", "top_pattern": ""}}""" 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 # ── 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], ) -> 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_result = compute_var(confidence=0.95, horizon_days=1, lookback_days=252, default_iv=0.20) 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": "", "context_driven": ["", ...], "general_knowledge": ["", ...], "key_signals": ["<3-5 signaux les plus décisifs>"], "context_log": {{ "macro_regime": "{dominant}", "geo_score": {geo_score_val}, "dominant_news": ["", "", ""], "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)} # ── 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, # 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] ], } return 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=90, 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": "", "winners_analysis": "", "losers_analysis": "", "key_lessons": ["", "", ""], "blind_spots": "", "next_cycle_priorities": "<3 priorités pour le prochain cycle>", "risk_watch": "<1-2 risques à surveiller>", "timing_note": "" }}""" 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, ) # Skip if last synthesis < 6 hours ago 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 < 6: logger.info( f"[AutoSynth] Super Contexte is {age_h:.1f}h old — skipping re-synthesis (threshold: 6h)" ) return except Exception: pass reports = list_ai_reports(limit=10) mtm_data = get_mtm_trades_with_traces(days=90) 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 _scheduler_loop(stop_event: threading.Event): """Background loop that runs the cycle at the configured interval.""" import time from services.database import get_config while not stop_event.is_set(): try: interval_hours = float(get_config("auto_cycle_hours") or "3") except Exception: interval_hours = 3.0 _current_status["interval_hours"] = interval_hours from datetime import timedelta next_run = (datetime.utcnow() + timedelta(hours=interval_hours)).isoformat() _current_status["next_run_at"] = next_run logger.info(f"[Scheduler] Next cycle in {interval_hours}h (at {next_run[:16]} UTC)") # Wait for the interval (or until stop is signalled) stop_event.wait(timeout=interval_hours * 3600) 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") 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 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 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, "last_cycle": last, "scheduler_alive": bool(_cycle_thread and _cycle_thread.is_alive()), }