diff --git a/backend/routers/cycle_actions.py b/backend/routers/cycle_actions.py index ee73e30..cabc830 100644 --- a/backend/routers/cycle_actions.py +++ b/backend/routers/cycle_actions.py @@ -1,39 +1,19 @@ """ -Isolated Cycle Actions — callable individually from UI or cron. +Isolated Cycle Actions — callable individually from UI or cron, to decompose +a full auto-cycle into steps runnable/inspectable one at a time. -Decomposition of the full auto_cycle into 8 independent actions: - - ACTION 1 — check-market-events ← IMPLEMENTED HERE - Sources: RSS news + FRED eco surprises + MA crossovers + institutional reports - Output : new market_events created in DB - - ACTION 2 — refresh-price-data (planned) - Sources: yfinance OHLCV for all watchlist instruments - Output : price_data table updated - - ACTION 3 — compute-indicators (planned) - Sources: price_data in DB - Output : ma20/50/100/200 + RSI + BB + ATR stored in instrument_indicators - - ACTION 4 — evaluate-impacts (planned) - Sources: market_events without impact scores - Output : GPT-evaluated impact_score + affected_assets stored - - ACTION 5 — generate-signals (planned) - Sources: indicators + market_events + portfolio - Output : trading signals for each instrument - - ACTION 6 — update-regime (planned) - Sources: signals + macro context - Output : current macro_regime classification - - ACTION 7 — snapshot-portfolio (planned) - Sources: portfolio positions + current prices - Output : risk metrics (VaR, PnL, delta) snapshot stored - - ACTION 8 — generate-report (planned) - Sources: all of the above - Output : daily cycle report PDF/JSON + ACTION 1 — check-market-events : RSS news + FRED surprises + MA crossovers + institutional reports -> new market_events + ACTION 2 — bootstrap-macro : seed historical macro/geopolitical events (idempotent) + ACTION 3 — bootstrap-eco : seed economic calendar events (idempotent) + ACTION 4 — bootstrap-categories : seed default impact categories (idempotent) + ACTION 5 — bootstrap-ma : historical MA50/100/200 crossover detection -> market_events (idempotent, >100 events skip unless force) + ACTION 6 — refresh-price-data : yfinance OHLCV for the watchlist -> price_data_cache + ACTION 7 — compute-indicators : MA/RSI/BB/ATR per watchlist ticker -> instrument_indicators + ACTION 8 — evaluate-impacts : AI-score market_events not yet in instrument_impacts + ACTION 9 — generate-signals : technical desk signals (check-market-events sources=[technical]) + wavelet scan + ACTION 10 — update-regime : macro gauges -> scenario scoring -> regime history -> cluster reclassification + ACTION 11 — snapshot-portfolio : VaR + PnL snapshot (same functions as /api/var/run-now + /api/var/pnl/run-now) + ACTION 12 — generate-report : best-effort standalone cycle report (see services.auto_cycle.generate_standalone_report) """ import logging from datetime import datetime @@ -122,14 +102,14 @@ def list_actions(): "implemented": True, "group": "bootstrap", }, - {"id": "bootstrap-ma", "label": "Bootstrap MA Crossovers", "description": "Detect historical MA50/100/200 crossovers for all instruments.", "status": "idle", "implemented": False, "group": "bootstrap"}, - {"id": "refresh-price-data", "label": "Refresh Price Data", "description": "Download fresh OHLCV for all watchlist instruments.", "status": "idle", "implemented": False, "group": "data"}, - {"id": "compute-indicators", "label": "Compute Indicators", "description": "Recalculate MA/RSI/BB/ATR for all instruments.", "status": "idle", "implemented": False, "group": "data"}, - {"id": "evaluate-impacts", "label": "Evaluate Impacts", "description": "AI-score market_events that lack impact_score.", "status": "idle", "implemented": False, "group": "ai"}, - {"id": "generate-signals", "label": "Generate Signals", "description": "Compute trading signals from indicators + events.", "status": "idle", "implemented": False, "group": "ai"}, - {"id": "update-regime", "label": "Update Regime", "description": "Re-classify the current macro regime.", "status": "idle", "implemented": False, "group": "ai"}, - {"id": "snapshot-portfolio", "label": "Snapshot Portfolio", "description": "Store a risk snapshot (VaR, PnL, delta).", "status": "idle", "implemented": False, "group": "portfolio"}, - {"id": "generate-report", "label": "Generate Report", "description": "Build and persist the daily cycle report.", "status": "idle", "implemented": False, "group": "portfolio"}, + {"id": "bootstrap-ma", "label": "Bootstrap MA Crossovers", "description": "Detect historical MA50/100/200 crossovers for all instruments.", "status": "running" if _running.get("bootstrap-ma") else "idle", "implemented": True, "group": "bootstrap"}, + {"id": "refresh-price-data", "label": "Refresh Price Data", "description": "Download fresh OHLCV for all watchlist instruments.", "status": "running" if _running.get("refresh-price-data") else "idle", "implemented": True, "group": "data"}, + {"id": "compute-indicators", "label": "Compute Indicators", "description": "Recalculate MA/RSI/BB/ATR for all instruments.", "status": "running" if _running.get("compute-indicators") else "idle", "implemented": True, "group": "data"}, + {"id": "evaluate-impacts", "label": "Evaluate Impacts", "description": "AI-score market_events that lack impact_score.", "status": "running" if _running.get("evaluate-impacts") else "idle", "implemented": True, "group": "ai"}, + {"id": "generate-signals", "label": "Generate Signals", "description": "Compute trading signals from indicators + events.", "status": "running" if _running.get("generate-signals") else "idle", "implemented": True, "group": "ai"}, + {"id": "update-regime", "label": "Update Regime", "description": "Re-classify the current macro regime.", "status": "running" if _running.get("update-regime") else "idle", "implemented": True, "group": "ai"}, + {"id": "snapshot-portfolio", "label": "Snapshot Portfolio", "description": "Store a risk snapshot (VaR, PnL, delta).", "status": "running" if _running.get("snapshot-portfolio") else "idle", "implemented": True, "group": "portfolio"}, + {"id": "generate-report", "label": "Generate Report", "description": "Build and persist the daily cycle report.", "status": "running" if _running.get("generate-report") else "idle", "implemented": True, "group": "portfolio"}, ] return {"actions": actions} @@ -267,3 +247,218 @@ def run_bootstrap_categories(force: bool = Query(False)): raise HTTPException(500, str(e)) _release("bootstrap-categories") return {"action": "bootstrap-categories", "status": "success", **result} + + +# ── POST /api/actions/bootstrap-ma ─────────────────────────────────────────── + +@router.post("/bootstrap-ma") +async def run_bootstrap_ma(force: bool = Query(False)): + """Detect historical MA50/100/200 crossovers for the 5 core underlyings. + Idempotent by default (skips if DB already has >100 market_events).""" + _guard("bootstrap-ma") + try: + from services.ma_analyzer import bootstrap_ma_events + result = await bootstrap_ma_events(force=force) + except Exception as e: + _release("bootstrap-ma") + raise HTTPException(500, str(e)) + _release("bootstrap-ma") + return {"action": "bootstrap-ma", "status": "success", **result} + + +# ── POST /api/actions/refresh-price-data ───────────────────────────────────── + +@router.post("/refresh-price-data") +def run_refresh_price_data(period: str = Query("3mo")): + """Download fresh OHLCV for every watchlist instrument into price_data_cache.""" + _guard("refresh-price-data") + try: + from services.price_cache import refresh_watchlist_price_data + result = refresh_watchlist_price_data(period=period) + except Exception as e: + _release("refresh-price-data") + raise HTTPException(500, str(e)) + _release("refresh-price-data") + return {"action": "refresh-price-data", "status": "success", **result} + + +# ── POST /api/actions/compute-indicators ───────────────────────────────────── + +@router.post("/compute-indicators") +def run_compute_indicators(horizon_days: int = Query(45)): + """Recalculate RSI/MA/Bollinger/ATR for every watchlist instrument, persisted + into instrument_indicators (calibration matches the horizon, same default + as the auto-cycle's own tech-indicators step).""" + _guard("compute-indicators") + try: + from services.technical_indicators import compute_and_save_indicators + result = compute_and_save_indicators(horizon_days=horizon_days) + except Exception as e: + _release("compute-indicators") + raise HTTPException(500, str(e)) + _release("compute-indicators") + return {"action": "compute-indicators", "status": "success", **result} + + +# ── POST /api/actions/evaluate-impacts ─────────────────────────────────────── + +@router.post("/evaluate-impacts") +def run_evaluate_impacts(limit: int = Query(20)): + """AI-score every market_event that has no row yet in instrument_impacts + (i.e. was never evaluated) — same per-event function as the manual + POST /api/impact/evaluate/bulk, with auto-discovered IDs instead of + caller-supplied ones.""" + _guard("evaluate-impacts") + try: + from services.database import get_conn + from services.impact_service import evaluate_event_impacts + conn = get_conn() + rows = conn.execute( + """SELECT id FROM market_events + WHERE id NOT IN (SELECT source_id FROM instrument_impacts WHERE source_type='event') + ORDER BY id DESC LIMIT ?""", + (limit,), + ).fetchall() + conn.close() + event_ids = [r["id"] for r in rows] + + evaluated = [] + failed = [] + for eid in event_ids: + try: + evaluate_event_impacts(eid, force=False) + evaluated.append(eid) + except Exception as e: + logger.warning(f"[action/evaluate-impacts] event {eid} failed: {e}") + failed.append(eid) + except Exception as e: + _release("evaluate-impacts") + raise HTTPException(500, str(e)) + _release("evaluate-impacts") + return { + "action": "evaluate-impacts", "status": "success", + "candidates_found": len(event_ids), "evaluated": len(evaluated), "failed": failed, + } + + +# ── POST /api/actions/generate-signals ─────────────────────────────────────── + +@router.post("/generate-signals") +def run_generate_signals(): + """Technical desk signals (ma_cross/rsi_extreme/bb_squeeze/new_52w_extreme, + via the same path as check-market-events sources=[technical]) + a fresh + wavelet band scan of the watchlist.""" + _guard("generate-signals") + try: + from datetime import datetime as _dt + from services.market_event_detector import check_new_market_events + from services.wavelet_signals import compute_and_save_wavelet_signals + + technical = check_new_market_events(sources=["technical"]) + run_id = _dt.utcnow().isoformat() + wavelet_results = compute_and_save_wavelet_signals(run_id) + except Exception as e: + _release("generate-signals") + raise HTTPException(500, str(e)) + _release("generate-signals") + return { + "action": "generate-signals", "status": "success", + "technical_events_created": technical.get("total_created", 0), + "wavelet_signals_computed": len(wavelet_results), + "wavelet_signals_fired": len([w for w in wavelet_results if w.get("signal_kind")]), + } + + +# ── POST /api/actions/update-regime ────────────────────────────────────────── + +@router.post("/update-regime") +def run_update_regime(): + """Re-fetch macro gauges, re-score scenarios, log the regime snapshot, and + re-run cluster reclassification (same sequence as auto_cycle's own step, + reading n_clusters/lookback_days from the configurable cycle_step_config).""" + _guard("update-regime") + try: + from services.data_fetcher import get_macro_gauges, score_macro_scenarios + from services.database import log_macro_regime + from services.database import detect_and_save_regime_clusters + from services.auto_cycle import get_cycle_step_config + + gauges = get_macro_gauges() + scenarios = score_macro_scenarios(gauges) + dominant = scenarios.get("dominant", "incertain") + 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, scenarios.get("scores", {}), scenarios.get("reasons", {}), gauges_summary) + + regime_cfg = get_cycle_step_config()["regime_clustering"] + cluster_result = None + if regime_cfg.get("enabled", True): + cluster_result = detect_and_save_regime_clusters( + n_clusters=regime_cfg["n_clusters"], days=regime_cfg["lookback_days"], + ) + except Exception as e: + _release("update-regime") + raise HTTPException(500, str(e)) + _release("update-regime") + return { + "action": "update-regime", "status": "success", + "dominant_regime": dominant, "scenario_scores": scenarios.get("scores", {}), + "cluster_result": cluster_result, + } + + +# ── POST /api/actions/snapshot-portfolio ───────────────────────────────────── + +@router.post("/snapshot-portfolio") +def run_snapshot_portfolio(): + """VaR + PnL snapshot — same functions already exposed individually at + POST /api/var/run-now and POST /api/var/pnl/run-now, combined into one + action for cycle-decomposition purposes.""" + _guard("snapshot-portfolio") + try: + from services.var_service import compute_var, save_var_snapshot, save_pnl_snapshot + from services.auto_cycle import get_cycle_step_config + + 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"], + ) + var_snapshot_id = None + if "error" not in var_result: + var_snapshot_id = save_var_snapshot( + var_result, confidence=var_cfg["confidence"], horizon_days=var_cfg["horizon_days"], + lookback_days=var_cfg["lookback_days"], default_iv=var_cfg["default_iv"], + ) + pnl_snapshot_id = save_pnl_snapshot() + except Exception as e: + _release("snapshot-portfolio") + raise HTTPException(500, str(e)) + _release("snapshot-portfolio") + return { + "action": "snapshot-portfolio", "status": "success", + "var_snapshot_id": var_snapshot_id, "pnl_snapshot_id": pnl_snapshot_id, + "var_result": var_result, + } + + +# ── POST /api/actions/generate-report ──────────────────────────────────────── + +@router.post("/generate-report") +def run_generate_report(): + """Best-effort standalone cycle report — see + services.auto_cycle.generate_standalone_report() for exactly which pieces + are DB-recovered vs. freshly recomputed vs. necessarily left empty + (artifacts that only exist inside a live run_cycle_once()).""" + _guard("generate-report") + try: + from services.auto_cycle import generate_standalone_report + result = generate_standalone_report() + except Exception as e: + _release("generate-report") + raise HTTPException(500, str(e)) + _release("generate-report") + return {"action": "generate-report", "status": "success", "run_id": result["run_id"]} diff --git a/backend/services/auto_cycle.py b/backend/services/auto_cycle.py index 42123ff..4fcd799 100644 --- a/backend/services/auto_cycle.py +++ b/backend/services/auto_cycle.py @@ -1754,6 +1754,79 @@ Réponds en JSON avec ce schéma EXACT: 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: diff --git a/backend/services/database.py b/backend/services/database.py index aa68c28..a497a8e 100644 --- a/backend/services/database.py +++ b/backend/services/database.py @@ -169,6 +169,23 @@ def init_db(): "ALTER TABLE wavelet_watchlist_signals ADD COLUMN energy REAL", "ALTER TABLE wavelet_watchlist_signals ADD COLUMN ridge_period_days REAL", "ALTER TABLE wavelet_watchlist_signals ADD COLUMN params_json TEXT", + # Cycle Actions — standalone "refresh-price-data" action, OHLCV cache + """CREATE TABLE IF NOT EXISTS price_data_cache ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ticker TEXT NOT NULL, + date TEXT NOT NULL, + open REAL, high REAL, low REAL, close REAL, volume REAL, + cached_at TEXT DEFAULT (datetime('now')), + UNIQUE(ticker, date) + )""", + # Cycle Actions — standalone "compute-indicators" action, one row per call per ticker + """CREATE TABLE IF NOT EXISTS instrument_indicators ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ticker TEXT NOT NULL, + computed_at TEXT DEFAULT (datetime('now')), + horizon_days INTEGER, + indicators_json TEXT DEFAULT '{}' + )""", ]: try: c.execute(_sql) @@ -183,6 +200,14 @@ def init_db(): c.execute("CREATE INDEX IF NOT EXISTS idx_atp_session_status ON ai_trade_proposals(session_id, status)") except Exception: pass + try: + c.execute("CREATE INDEX IF NOT EXISTS idx_pdc_ticker_date ON price_data_cache(ticker, date DESC)") + except Exception: + pass + try: + c.execute("CREATE INDEX IF NOT EXISTS idx_ii_ticker_date ON instrument_indicators(ticker, computed_at DESC)") + except Exception: + pass try: c.execute("CREATE INDEX IF NOT EXISTS idx_chat_session_date ON ai_chat_messages(session_id, created_at)") @@ -3441,6 +3466,72 @@ def resolve_ai_trade_proposal(proposal_id: str, status: str, portfolio_id: Optio conn.close() +# ── Cycle Actions — standalone price data cache + indicator snapshots ────────── + +def upsert_price_data(ticker: str, rows: List[Dict[str, Any]]) -> int: + """rows: [{date, open, high, low, close, volume}, ...]. Returns rows written.""" + if not rows: + return 0 + conn = get_conn() + n = 0 + for r in rows: + conn.execute( + """INSERT INTO price_data_cache (ticker, date, open, high, low, close, volume, cached_at) + VALUES (?, ?, ?, ?, ?, ?, ?, datetime('now')) + ON CONFLICT(ticker, date) DO UPDATE SET + open=excluded.open, high=excluded.high, low=excluded.low, + close=excluded.close, volume=excluded.volume, cached_at=excluded.cached_at""", + (ticker.upper(), r.get("date"), r.get("open"), r.get("high"), r.get("low"), r.get("close"), r.get("volume")), + ) + n += 1 + conn.commit() + conn.close() + return n + + +def get_cached_price_data(ticker: str, limit: int = 90) -> List[Dict[str, Any]]: + conn = get_conn() + rows = conn.execute( + "SELECT * FROM price_data_cache WHERE ticker=? ORDER BY date DESC LIMIT ?", + (ticker.upper(), limit), + ).fetchall() + conn.close() + return [dict(r) for r in rows] + + +def save_instrument_indicators(ticker: str, horizon_days: int, indicators: Dict[str, Any]) -> None: + conn = get_conn() + conn.execute( + "INSERT INTO instrument_indicators (ticker, horizon_days, indicators_json) VALUES (?, ?, ?)", + (ticker.upper(), horizon_days, json.dumps(indicators)), + ) + conn.commit() + conn.close() + + +def get_latest_instrument_indicators() -> List[Dict[str, Any]]: + """Most recent indicators row per ticker.""" + conn = get_conn() + rows = conn.execute( + """SELECT i.* FROM instrument_indicators i + INNER JOIN ( + SELECT ticker, MAX(computed_at) AS max_computed_at + FROM instrument_indicators GROUP BY ticker + ) latest ON i.ticker = latest.ticker AND i.computed_at = latest.max_computed_at + ORDER BY i.ticker""" + ).fetchall() + conn.close() + out = [] + for r in rows: + d = dict(r) + try: + d["indicators"] = json.loads(d.pop("indicators_json") or "{}") + except Exception: + d["indicators"] = {} + out.append(d) + return out + + # ── System Logs ─────────────────────────────────────────────────────────────── def log_system_event( diff --git a/backend/services/ma_analyzer.py b/backend/services/ma_analyzer.py index a1af4d9..1076bf4 100644 --- a/backend/services/ma_analyzer.py +++ b/backend/services/ma_analyzer.py @@ -495,14 +495,15 @@ def _to_event_dict(ev: Dict[str, Any]) -> Dict[str, Any]: } -async def bootstrap_ma_events() -> Dict[str, Any]: +async def bootstrap_ma_events(force: bool = False) -> Dict[str, Any]: """ Main entrypoint. Detects MA ruptures, enriches with GPT, saves to DB. - Idempotent: skips if DB already has > 100 events. + Idempotent by default: skips if DB already has > 100 events. Pass force=True + to bypass this guard (same convention as bootstrap_macro_events/bootstrap_eco_events). Returns {"detected": N, "saved": N, "skipped": N}. """ existing = count_market_events() - if existing > 100: + if existing > 100 and not force: logger.info(f"[MA] DB already has {existing} events — skipping bootstrap") return {"detected": 0, "saved": 0, "skipped": existing} diff --git a/backend/services/price_cache.py b/backend/services/price_cache.py new file mode 100644 index 0000000..8e237d8 --- /dev/null +++ b/backend/services/price_cache.py @@ -0,0 +1,41 @@ +""" +Cycle Actions — standalone "refresh-price-data" action. + +Downloads fresh OHLCV for every watchlist instrument and caches it in +price_data_cache. Deliberately independent from the technical-indicators / +wavelet code paths (which each do their own live yfinance fetch) — this is an +inspection/refresh tool for decomposing a cycle, not a shared cache other +steps depend on, so it can't regress the already-tested live cycle path. +""" +import logging +from typing import Any, Dict + +logger = logging.getLogger(__name__) + + +def refresh_watchlist_price_data(period: str = "3mo") -> Dict[str, Any]: + from services.database import get_instruments_watchlist, upsert_price_data + from services.data_fetcher import get_historical + + tickers = [w["ticker"] for w in get_instruments_watchlist()] + per_ticker: Dict[str, int] = {} + failed = [] + + for ticker in tickers: + try: + rows = get_historical(ticker, period=period, interval="1d") + if not rows: + failed.append(ticker) + continue + n = upsert_price_data(ticker, rows) + per_ticker[ticker] = n + except Exception as e: + logger.warning(f"[PriceCache] Failed to refresh {ticker}: {e}") + failed.append(ticker) + + return { + "tickers_refreshed": len(per_ticker), + "rows_written": sum(per_ticker.values()), + "per_ticker": per_ticker, + "failed": failed, + } diff --git a/backend/services/technical_indicators.py b/backend/services/technical_indicators.py index e231b30..6f0df94 100644 --- a/backend/services/technical_indicators.py +++ b/backend/services/technical_indicators.py @@ -175,3 +175,23 @@ def format_indicators_for_prompt(indicators: dict) -> str: if "error" in indicators: return "" return indicators.get("prompt_block", "") + + +def compute_and_save_indicators(horizon_days: int = 45) -> dict: + """Cycle Actions — standalone "compute-indicators" action. compute_indicators() + itself is pure (no persistence) — this loops the watchlist and persists each + result into instrument_indicators (which nothing else reads from yet; this is + an inspection snapshot, not a cache other steps depend on).""" + from services.database import get_instruments_watchlist, save_instrument_indicators + + tickers = [w["ticker"] for w in get_instruments_watchlist()] + computed = 0 + failed = [] + for ticker in tickers: + result = compute_indicators(ticker, horizon_days=horizon_days) + if "error" in result: + failed.append(ticker) + continue + save_instrument_indicators(ticker, horizon_days, result) + computed += 1 + return {"tickers_computed": computed, "failed": failed, "horizon_days": horizon_days}