Files
OpenFin/backend/routers/cycle.py
OpenSquared 2d474c9194 feat: cycle
2026-07-15 12:03:02 +02:00

259 lines
11 KiB
Python

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import Any, Dict, Optional
from services.database import get_cycle_runs, get_cycle_run, set_config, get_config, list_cycle_context_snapshots, get_cycle_context_snapshot, get_ai_call_logs
from services.auto_cycle import get_status, trigger_manual, restart_scheduler, CYCLE_STEP_CATALOG
router = APIRouter(prefix="/api/cycle", tags=["cycle"])
@router.get("/status")
def cycle_status():
"""Current scheduler state + last cycle summary."""
return get_status()
@router.get("/step-catalog")
def cycle_step_catalog():
"""Self-describing catalog of every configurable cycle step (mirrors
GET /api/ai-desks/signal-catalog) — lets Config.tsx render a generic
form instead of hand-coded sliders per knob."""
return {"steps": CYCLE_STEP_CATALOG}
@router.get("/history")
def cycle_history(limit: int = 20):
"""List recent cycle runs."""
runs = get_cycle_runs(limit=limit)
# Parse commentary JSON for frontend
import json
for r in runs:
if r.get("commentary"):
try:
r["commentary_parsed"] = json.loads(r["commentary"])
except Exception:
r["commentary_parsed"] = {"commentary": r["commentary"]}
return {"runs": runs, "count": len(runs)}
@router.post("/trigger")
def trigger_cycle():
"""Manually trigger one cycle immediately (non-blocking)."""
key = get_config("openai_api_key") or ""
if not key:
raise HTTPException(400, "Clé OpenAI non configurée")
trigger_manual()
return {"triggered": True, "message": "Cycle lancé en arrière-plan"}
class CycleConfigRequest(BaseModel):
enabled: Optional[bool] = None
interval_hours: Optional[float] = None
similarity_threshold: Optional[float] = None
min_ev_threshold: Optional[float] = None
min_score_threshold: Optional[int] = None
trade_budget_eur: Optional[float] = None
preferred_horizon_min: Optional[int] = None
preferred_horizon_max: Optional[int] = None
journal_retention_days: Optional[int] = None
maturity_threshold_pct: Optional[int] = None
weekend_cycle_enabled: Optional[bool] = None
weekend_cycle_times: Optional[str] = None # "HH:MM,HH:MM" in UTC
cycle_step_config: Optional[Dict[str, Dict[str, Any]]] = None # {step_id: {param: value}}
@router.post("/config")
def update_cycle_config(req: CycleConfigRequest):
"""Update auto-cycle + filter configuration and restart scheduler."""
if req.enabled is not None:
set_config("auto_cycle_enabled", "true" if req.enabled else "false")
if req.interval_hours is not None:
if not (0.5 <= req.interval_hours <= 24):
raise HTTPException(400, "interval_hours must be between 0.5 and 24")
set_config("auto_cycle_hours", str(req.interval_hours))
if req.similarity_threshold is not None:
if not (0.0 <= req.similarity_threshold <= 1.0):
raise HTTPException(400, "similarity_threshold must be between 0 and 1")
set_config("auto_cycle_similarity_threshold", str(req.similarity_threshold))
if req.min_ev_threshold is not None:
if not (0.0 <= req.min_ev_threshold <= 1.0):
raise HTTPException(400, "min_ev_threshold must be between 0 and 1")
set_config("min_ev_threshold", str(req.min_ev_threshold))
if req.min_score_threshold is not None:
if not (0 <= req.min_score_threshold <= 100):
raise HTTPException(400, "min_score_threshold must be between 0 and 100")
set_config("min_score_threshold", str(req.min_score_threshold))
if req.journal_retention_days is not None:
if not (7 <= req.journal_retention_days <= 365):
raise HTTPException(400, "journal_retention_days must be between 7 and 365")
set_config("journal_retention_days", str(req.journal_retention_days))
if req.maturity_threshold_pct is not None:
if not (5 <= req.maturity_threshold_pct <= 75):
raise HTTPException(400, "maturity_threshold_pct must be between 5 and 75")
set_config("maturity_threshold_pct", str(req.maturity_threshold_pct))
if req.weekend_cycle_enabled is not None:
set_config("weekend_cycle_enabled", "true" if req.weekend_cycle_enabled else "false")
if req.trade_budget_eur is not None:
if not (100 <= req.trade_budget_eur <= 500_000):
raise HTTPException(400, "trade_budget_eur must be between 100 and 500000")
set_config("trade_budget_eur", str(req.trade_budget_eur))
if req.preferred_horizon_min is not None:
if not (1 <= req.preferred_horizon_min <= 365):
raise HTTPException(400, "preferred_horizon_min must be between 1 and 365")
set_config("preferred_horizon_min", str(req.preferred_horizon_min))
if req.preferred_horizon_max is not None:
if not (1 <= req.preferred_horizon_max <= 365):
raise HTTPException(400, "preferred_horizon_max must be between 1 and 365")
set_config("preferred_horizon_max", str(req.preferred_horizon_max))
if req.weekend_cycle_times is not None:
# Validate format: comma-separated HH:MM values
import re
if not re.match(r'^(\d{2}:\d{2})(,\d{2}:\d{2})*$', req.weekend_cycle_times.strip()):
raise HTTPException(400, "weekend_cycle_times must be 'HH:MM' or 'HH:MM,HH:MM,...'")
set_config("weekend_cycle_times", req.weekend_cycle_times.strip())
if req.cycle_step_config is not None:
import json
known_ids = {step["id"] for step in CYCLE_STEP_CATALOG}
unknown = set(req.cycle_step_config.keys()) - known_ids
if unknown:
raise HTTPException(400, f"Unknown cycle step id(s): {sorted(unknown)}")
# Merge over whatever is already saved so a partial update from the UI
# never wipes out other steps' settings.
current = json.loads(get_config("cycle_step_config") or "{}")
for step_id, params in req.cycle_step_config.items():
current[step_id] = {**current.get(step_id, {}), **params}
set_config("cycle_step_config", json.dumps(current))
# Restart scheduler to pick up changes
restart_scheduler()
return get_status()
@router.get("/contexts")
def list_context_snapshots(limit: int = 30):
"""List cycle context snapshots (most recent first)."""
return {"snapshots": list_cycle_context_snapshots(limit=limit)}
def _sanitize_floats(obj):
"""Recursively replace NaN/Inf floats with None for JSON-safe serialization."""
import math
if isinstance(obj, float):
return None if (math.isnan(obj) or math.isinf(obj)) else obj
if isinstance(obj, dict):
return {k: _sanitize_floats(v) for k, v in obj.items()}
if isinstance(obj, list):
return [_sanitize_floats(v) for v in obj]
return obj
@router.get("/contexts/{run_id}")
def get_context_snapshot(run_id: str):
"""Return the full context snapshot for a given cycle run_id."""
snap = get_cycle_context_snapshot(run_id)
if not snap:
raise HTTPException(404, "Snapshot non trouvé pour ce cycle")
return _sanitize_floats(snap)
@router.get("/ai-calls/{run_id}")
def get_cycle_ai_calls(run_id: str):
"""Return all AI calls logged for a given cycle run_id."""
calls = get_ai_call_logs(run_id)
return {"run_id": run_id, "calls": calls, "count": len(calls)}
class ReplayRequest(BaseModel):
override_notes: Optional[str] = None # optional annotation added to the replay
@router.post("/contexts/{run_id}/replay")
def replay_cycle(run_id: str, req: ReplayRequest):
"""
Phase 5 — Re-run AI suggestion with the historical context from a saved snapshot.
Returns new pattern suggestions based on the original context data.
"""
snap = get_cycle_context_snapshot(run_id)
if not snap:
raise HTTPException(404, "Snapshot non trouvé pour ce cycle")
ctx = snap.get("context", {})
if not ctx:
raise HTTPException(400, "Snapshot vide — impossible de rejouer")
try:
from services.ai_analyzer import suggest_patterns_from_market_context, apply_news_decay, partition_news_by_age
import json
# Reconstruct minimal inputs from the snapshot
# news: merge inter_cycle + recent_24h + older from partitioned snapshot
news_part = ctx.get("news_partitioned", {})
news_flat = (
news_part.get("inter_cycle", [])
+ news_part.get("recent_24h", [])
+ news_part.get("older", [])
)
# quotes: reconstruct from quotes_summary
quotes_by_class = ctx.get("quotes_summary", {})
# calendar
calendar = ctx.get("calendar", [])
# macro regime (simplified for replay)
macro_regime = None
if ctx.get("macro_regime"):
macro_regime = {"scenarios": {"dominant": ctx["macro_regime"].get("dominant"), "scores": ctx["macro_regime"].get("scores"), "asset_bias": {}, "reasons": []}, "gauges": {}}
# geo score
geo_score = ctx.get("geo_score", {"score": 50, "level": "medium", "top_risks": []})
# preserved blocks
tech_block = ctx.get("tech_indicators_block", "")
iv_context = ctx.get("iv_context_preview", "")
cycle_meta = ctx.get("cycle_meta", {})
# Rebuild FRED block from saved releases
fred_block = ""
if ctx.get("fred_releases"):
from services.fred_fetcher import build_fred_context_block
fred_block = build_fred_context_block(ctx["fred_releases"])
# Rebuild price discovery block from saved absorptions
pd_block = ""
if ctx.get("price_discovery"):
from services.price_discovery import build_price_discovery_block
pd_block = build_price_discovery_block(ctx["price_discovery"])
# Add replay note to cycle_meta
if req.override_notes:
cycle_meta = {**cycle_meta, "replay_notes": req.override_notes}
cycle_meta["is_replay"] = True
cycle_meta["replayed_at"] = __import__("datetime").datetime.utcnow().isoformat()
suggestions = suggest_patterns_from_market_context(
news=news_flat,
quotes_by_class=quotes_by_class,
calendar=calendar,
macro_regime=macro_regime,
geo_score=geo_score,
iv_context=iv_context,
cycle_meta=cycle_meta,
tech_indicators_block=tech_block,
fred_block=fred_block,
price_discovery_block=pd_block,
)
return {
"run_id": run_id,
"original_ts": snap["ts"],
"replayed_at": cycle_meta["replayed_at"],
"override_notes": req.override_notes,
"suggestions_count": len(suggestions),
"suggestions": suggestions,
}
except Exception as e:
raise HTTPException(500, f"Replay failed: {str(e)}")