224 lines
9.4 KiB
Python
224 lines
9.4 KiB
Python
"""
|
|
Context assembly for the free-form AI chat widget. Every block below reuses an
|
|
existing, independent builder already used by the auto-cycle prompt (see
|
|
services/auto_cycle.py Step 1.9/2) — nothing here duplicates that logic, it just
|
|
re-packages the same read-only data for an interactive Q&A session instead of a
|
|
decision-making cycle. No block writes to the DB or triggers anything.
|
|
"""
|
|
import time
|
|
from typing import Dict, List
|
|
|
|
CONTEXT_BLOCKS = [
|
|
"portfolio", "geo_news", "macro", "patterns", "options_iv", "tech_indicators",
|
|
"wavelet_signals", "watchlist_quotes", "economic_calendar", "institutional",
|
|
"super_context", "var_risk",
|
|
]
|
|
|
|
_CACHE_TTL_SECONDS = 10 * 60
|
|
_context_cache: Dict[str, Dict] = {} # session_id -> {"blocks": {...}, "ts": float}
|
|
|
|
|
|
def _block_portfolio() -> str:
|
|
from services.portfolio_context import get_open_trades_with_moves, get_portfolio_concentration, build_portfolio_context_block
|
|
trades = get_open_trades_with_moves()
|
|
conc = get_portfolio_concentration(trades)
|
|
return build_portfolio_context_block(trades, conc)
|
|
|
|
|
|
def _block_geo_news() -> str:
|
|
from services.data_fetcher import fetch_geo_news
|
|
from services.geo_analyzer import compute_geo_risk_score
|
|
news = fetch_geo_news()
|
|
score = compute_geo_risk_score(news)
|
|
top = sorted(news, key=lambda n: -(n.get("impact_score") or 0))[:12]
|
|
lines = [f"## GEOPOLITICAL RISK\nScore: {score['score']}/100 ({score['level']})", "Top news:"]
|
|
for n in top:
|
|
lines.append(f"- [{round((n.get('impact_score') or 0) * 100)}] {n.get('title')}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _block_macro() -> str:
|
|
from services.data_fetcher import get_macro_gauges, score_macro_scenarios
|
|
gauges = get_macro_gauges()
|
|
scenarios = score_macro_scenarios(gauges)
|
|
lines = [f"## MACRO REGIME\nDominant scenario: {scenarios.get('dominant')}"]
|
|
ranked = scenarios.get("ranked") or []
|
|
if ranked:
|
|
lines.append("Scenario scores: " + ", ".join(f"{k}={v}" for k, v in ranked[:5]))
|
|
lines.append("Key gauges:")
|
|
for gid, g in list(gauges.items())[:12]:
|
|
lines.append(f"- {g.get('label', gid)}: {g.get('value')} {g.get('unit', '')} ({g.get('change_pct')}%)")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _block_patterns() -> str:
|
|
from services.database import get_all_pattern_reliability_map
|
|
rel = get_all_pattern_reliability_map()
|
|
if not rel:
|
|
return "## PATTERN RELIABILITY\nNo data yet."
|
|
lines = ["## PATTERN RELIABILITY"]
|
|
for pid, stats in list(rel.items())[:15]:
|
|
lines.append(f"- {stats.get('pattern_id', pid)}: win_rate={stats.get('win_rate')}, trades={stats.get('trade_count')}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _block_options_iv() -> str:
|
|
from services.database import get_instruments_watchlist
|
|
from services.iv_engine import get_iv_context_for_prompt
|
|
tickers = [w["ticker"] for w in get_instruments_watchlist()]
|
|
if not tickers:
|
|
return "## OPTIONS / IV\nNo watchlist instruments configured (Config > Watchlist)."
|
|
return "## OPTIONS / IV (watchlist)\n" + get_iv_context_for_prompt(tickers)
|
|
|
|
|
|
def _block_tech_indicators() -> str:
|
|
from services.database import get_instruments_watchlist
|
|
from services.technical_indicators import compute_indicators, format_indicators_for_prompt
|
|
tickers = [w["ticker"] for w in get_instruments_watchlist()][:8]
|
|
lines = ["## TECHNICAL INDICATORS (watchlist)"]
|
|
for t in tickers:
|
|
try:
|
|
block = format_indicators_for_prompt(compute_indicators(t, horizon_days=45))
|
|
except Exception:
|
|
block = ""
|
|
if block:
|
|
lines.append(f"### {t}\n{block}")
|
|
return "\n".join(lines) if len(lines) > 1 else "## TECHNICAL INDICATORS\nNo data available."
|
|
|
|
|
|
def _block_wavelet_signals() -> str:
|
|
from services.database import get_latest_wavelet_state
|
|
rows = get_latest_wavelet_state()
|
|
if not rows:
|
|
return "## WAVELET SIGNALS\nNo wavelet state computed yet (computed each auto-cycle for the watchlist instruments)."
|
|
|
|
by_ticker: Dict[str, List[Dict]] = {}
|
|
for r in rows:
|
|
by_ticker.setdefault(r["ticker"], []).append(r)
|
|
|
|
lines = ["## WAVELET SIGNALS (watchlist, latest cycle — slope/energy/ridge state + any active trigger)"]
|
|
for ticker, band_rows in list(by_ticker.items())[:12]:
|
|
lines.append(f"### {ticker}")
|
|
for r in band_rows:
|
|
tag = f" -> SIGNAL {r['signal_kind']} ({r['direction']})" if r.get("signal_kind") else ""
|
|
if r["band_label"] == "ridge":
|
|
if r.get("ridge_period_days") is not None:
|
|
lines.append(f"- ridge (cycle dominant): {r['ridge_period_days']:.1f}j{tag}")
|
|
continue
|
|
period = f"{r['period_low_days']}-{r['period_high_days']}j" if r.get("period_low_days") is not None else r["band_label"]
|
|
slope = r.get("slope")
|
|
slope_txt = f"pente {'+' if slope >= 0 else ''}{slope:.4f}" if slope is not None else "pente n/a"
|
|
energy_txt = f", energie {r['energy']:.4f}" if r.get("energy") is not None else ""
|
|
lines.append(f"- {r['band_label']} [{period}]: valeur {r.get('value')}, {slope_txt}{energy_txt}{tag}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _block_watchlist_quotes() -> str:
|
|
from services.database import get_instruments_watchlist
|
|
from services.data_fetcher import get_quote
|
|
items = get_instruments_watchlist()
|
|
if not items:
|
|
return "## WATCHLIST\nEmpty — no instruments configured."
|
|
lines = ["## WATCHLIST QUOTES"]
|
|
for w in items:
|
|
q = get_quote(w["ticker"]) or {}
|
|
lines.append(f"- {w['ticker']} ({w.get('asset_class')}): {q.get('price')} ({q.get('change_pct')}%)")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _block_economic_calendar() -> str:
|
|
from services.ff_calendar import get_calendar
|
|
data = get_calendar(period="recent", limit=50)
|
|
events = [e for e in data.get("events", []) if e.get("impact") in ("high", "medium")]
|
|
if not events:
|
|
return "## ECONOMIC CALENDAR\nNo high/medium impact events in range."
|
|
lines = ["## ECONOMIC CALENDAR (high/medium impact, recent window)"]
|
|
for e in events[:15]:
|
|
lines.append(f"- {e['event_date']} {e.get('event_time') or ''} [{e['currency']}] {e['event_name']} ({e['impact']})")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _block_institutional() -> str:
|
|
from services.ai_analyzer import build_institutional_block
|
|
return "## INSTITUTIONAL REPORTS\n" + build_institutional_block(days=7)
|
|
|
|
|
|
def _block_super_context() -> str:
|
|
from services.database import get_latest_portfolio_lessons, get_latest_reasoning_state
|
|
lessons = get_latest_portfolio_lessons()
|
|
reasoning = get_latest_reasoning_state()
|
|
lines = ["## SUPER CONTEXT / LESSONS LEARNED"]
|
|
if lessons:
|
|
lines.append(f"Headline: {lessons.get('headline', '')}")
|
|
key_lessons = lessons.get("key_lessons") or []
|
|
if key_lessons:
|
|
lines.append("Key lessons: " + "; ".join(str(k) for k in key_lessons[:5]))
|
|
if lessons.get("risk_watch"):
|
|
lines.append(f"Risk watch: {lessons['risk_watch']}")
|
|
if reasoning and reasoning.get("narrative"):
|
|
lines.append(f"Reasoning state (v{reasoning.get('version')}): {reasoning['narrative'][:600]}")
|
|
if len(lines) == 1:
|
|
return "## SUPER CONTEXT\nNo data yet."
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _block_var_risk() -> str:
|
|
from services.var_service import get_latest_var_snapshot
|
|
from services.portfolio_risk import analyze_simulation_portfolio, build_monitor_context
|
|
lines = ["## RISK / VaR"]
|
|
snap = get_latest_var_snapshot()
|
|
if snap:
|
|
lines.append(
|
|
f"VaR 95% hist: {snap.get('hist_var_1d_pct')}% · CVaR: {snap.get('hist_cvar_pct')}% · "
|
|
f"Monte Carlo x1.5: {snap.get('mc_var_1d_pct')}%"
|
|
)
|
|
else:
|
|
lines.append("No VaR snapshot yet.")
|
|
try:
|
|
risk = analyze_simulation_portfolio()
|
|
lines.append(build_monitor_context(risk))
|
|
except Exception:
|
|
pass
|
|
return "\n".join(lines)
|
|
|
|
|
|
_BLOCK_BUILDERS = {
|
|
"portfolio": _block_portfolio,
|
|
"geo_news": _block_geo_news,
|
|
"macro": _block_macro,
|
|
"patterns": _block_patterns,
|
|
"options_iv": _block_options_iv,
|
|
"tech_indicators": _block_tech_indicators,
|
|
"wavelet_signals": _block_wavelet_signals,
|
|
"watchlist_quotes": _block_watchlist_quotes,
|
|
"economic_calendar": _block_economic_calendar,
|
|
"institutional": _block_institutional,
|
|
"super_context": _block_super_context,
|
|
"var_risk": _block_var_risk,
|
|
}
|
|
|
|
|
|
def assemble_context(enabled_blocks: List[str], session_id: str, refresh: bool = False) -> Dict[str, str]:
|
|
"""Return {block_key: formatted_text} for every requested, known block.
|
|
Cached in-memory per session (same TTL-cache idiom as _macro_cache in
|
|
routers/market_data.py) so a back-and-forth conversation doesn't re-run every
|
|
builder (some hit yfinance) on every single message."""
|
|
cached = _context_cache.get(session_id)
|
|
if not refresh and cached and (time.time() - cached["ts"]) < _CACHE_TTL_SECONDS:
|
|
blocks = cached["blocks"]
|
|
else:
|
|
blocks = {}
|
|
for key in CONTEXT_BLOCKS:
|
|
try:
|
|
blocks[key] = _BLOCK_BUILDERS[key]()
|
|
except Exception as e:
|
|
blocks[key] = f"## {key.upper()}\n(unavailable: {e})"
|
|
_context_cache[session_id] = {"blocks": blocks, "ts": time.time()}
|
|
|
|
requested = [b for b in enabled_blocks if b in blocks] or list(blocks.keys())
|
|
return {k: blocks[k] for k in requested}
|
|
|
|
|
|
def clear_context_cache(session_id: str) -> None:
|
|
_context_cache.pop(session_id, None)
|