admin/OpenFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Phase 1 — delta temporel + decay news + cycle_meta dans prompts IA

Browse Source 
- database.py: get_last_completed_cycle_ts() pour mesurer le delta entre cycles
- auto_cycle.py: calcul delta_minutes + cycle_meta dict transmis aux fonctions IA
- ai_analyzer.py: apply_news_decay() (halflife par catégorie), partition_news_by_age()
  (3 buckets: inter_cycle / recent_24h / older), _build_temporal_news_block() pour
  le prompt suggestion; cycle_meta injecté aussi dans score_patterns_with_context()

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OpenSquared
2026-06-20 16:36:54 +02:00
parent fff0c15316
commit d5e31bc897
3 changed files with 211 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

169
backend/services/ai_analyzer.py
View File

@@ -6,6 +6,8 @@ from openai import OpenAI
from typing import Optional, List, Dict, Any
import json
import os
import math
from datetime import datetime, timezone
_client: Optional[OpenAI] = None
@@ -350,6 +352,7 @@ def score_patterns_with_context(
portfolio_lessons: Optional[Dict] = None,
iv_context: str = "",
risk_context: str = "",
cycle_meta: Optional[Dict] = None,
) -> List[Dict[str, Any]]:
"""Score all patterns with rich context (news, prices, IV, risk clusters) using GPT-4o."""
if not get_client():
@@ -550,9 +553,28 @@ Instructions de notation:
- Indique dans "summary": [GOLDILOCKS|STAGFLATION|RÉCESSION|DÉSINFLATION|CRISE] + [SUPPORTING|NEUTRAL|CONTRA]
"""
# Temporal context block for scoring
_cm = cycle_meta or {}
_delta_min_sc = _cm.get("delta_minutes", 180)
_calib_sc = _cm.get("calibration_label", "")
temporal_section_sc = ""
if _cm:
# Apply decay to news used in scoring
news_decayed_sc = apply_news_decay(recent_news)
_partitioned_sc = partition_news_by_age(news_decayed_sc, _delta_min_sc)
_inter_sc = _partitioned_sc["inter_cycle"]
temporal_section_sc = (
f"\nCONTEXTE TEMPOREL DU CYCLE:\n"
f"- Dernier cycle il y a : {_delta_min_sc:.0f} minutes | {_calib_sc}\n"
f"- NEWS INTER-CYCLE (non encore pricées, {len(_inter_sc)} articles) : "
+ " | ".join(n.get("title", "")[:60] for n in _inter_sc[:3]) + "\n"
f"⚠️ Tiens compte du délai depuis le dernier cycle pour évaluer si les news sont déjà intégrées.\n"
)
user = f"""CONTEXTE GLOBAL:
- Score risque géopolitique: {geo_score.get('score', 50)}/100 ({geo_score.get('level', 'medium')})
- Top risques: {geo_score.get('top_risks', [])}
{temporal_section_sc}
{macro_section}
TEMPLATE DE NOTATION:
{scoring_template}
@@ -810,6 +832,128 @@ TEMPLATE DE NOTATION:
# ── Suggest new patterns from live market context ─────────────────────────────
# ── Temporal news utilities ───────────────────────────────────────────────────
# Halflife par catégorie de news (heures) — après combien de temps l'impact est réduit de moitié
_NEWS_HALFLIFE_HOURS: Dict[str, float] = {
"economic": 4.0, # CPI, NFP, PIB, FOMC → pricés très vite
"geopolitical": 48.0, # conflit, sanction, accord → digestion lente
"corporate": 12.0, # earnings, deal, fusion → digestion moyenne
"default": 24.0,
}
_ECONOMIC_KEYWORDS = {"cpi", "inflation", "nfp", "jobs", "fomc", "fed", "gdp", "pmi", "ecb", "bce", "earnings", "taux"}
_GEO_KEYWORDS = {"war", "guerre", "sanction", "conflit", "ceasefire", "accord", "treaty", "nuclear", "missile", "invasion", "trump", "israel", "ukraine", "russia", "iran", "china"}
def _news_category(title: str) -> str:
t = title.lower()
if any(k in t for k in _ECONOMIC_KEYWORDS):
return "economic"
if any(k in t for k in _GEO_KEYWORDS):
return "geopolitical"
return "corporate"
def _article_age_hours(article: Dict) -> float:
"""Return age of article in hours from now. Falls back to 6h if no timestamp."""
for field in ("published", "published_at", "fetched_at"):
raw = article.get(field, "")
if not raw:
continue
try:
# Strip timezone info for naive comparison
raw_clean = str(raw).replace("Z", "+00:00")
dt = datetime.fromisoformat(raw_clean)
if dt.tzinfo is not None:
dt = dt.replace(tzinfo=None) - dt.utcoffset()
age = (datetime.utcnow() - dt).total_seconds() / 3600
return max(0.0, age)
except Exception:
continue
return 6.0 # default: assume 6h old
def apply_news_decay(news: List[Dict]) -> List[Dict]:
"""
Add `decayed_score` to each article: impact_score × exp(-age / halflife).
Does NOT modify original list — returns new list with added field.
"""
result = []
for n in news:
cat = _news_category(n.get("title", ""))
halflife = _NEWS_HALFLIFE_HOURS.get(cat, _NEWS_HALFLIFE_HOURS["default"])
age_h = _article_age_hours(n)
raw_score = float(n.get("impact_score") or 0)
decayed = raw_score * math.exp(-age_h / halflife)
result.append({**n, "decayed_score": round(decayed, 4), "age_hours": round(age_h, 1), "news_category": cat})
return result
def partition_news_by_age(news: List[Dict], delta_minutes: float) -> Dict[str, List[Dict]]:
"""
Split news into temporal buckets relative to cycle delta:
- inter_cycle : published within delta_minutes → potentiellement PAS ENCORE dans les prix
- recent_24h : published 24h → delta_minutes ago → partiellement pricés
- older : > 24h → considérés comme pricés par le marché
"""
delta_hours = delta_minutes / 60.0
inter, recent, older = [], [], []
for n in news:
age_h = n.get("age_hours", _article_age_hours(n))
if age_h <= delta_hours:
inter.append(n)
elif age_h <= 24.0:
recent.append(n)
else:
older.append(n)
# Sort each bucket by decayed_score desc
key = lambda x: -(x.get("decayed_score") or x.get("impact_score") or 0)
return {
"inter_cycle": sorted(inter, key=key),
"recent_24h": sorted(recent, key=key),
"older": sorted(older, key=key),
}
def _build_temporal_news_block(partitioned: Dict[str, List], cycle_meta: Dict) -> str:
"""Build the news section of the IA prompt with temporal framing."""
delta_min = cycle_meta.get("delta_minutes", 180)
calib = cycle_meta.get("calibration_label", "")
def _fmt_news(articles: List[Dict], limit: int = 6) -> str:
lines = []
for n in articles[:limit]:
score = n.get("decayed_score") or n.get("impact_score") or 0
age = n.get("age_hours", "?")
lines.append(
f" • [{n.get('source','')}] {n.get('title','')[:110]}"
f" (impact={score:.2f}, âge={age:.0f}h)"
)
return "\n".join(lines) if lines else " (aucune)"
inter = partitioned.get("inter_cycle", [])
recent = partitioned.get("recent_24h", [])
older = partitioned.get("older", [])
block = f"""
## ⏱ CONTEXTE TEMPOREL DU CYCLE
- Dernier cycle il y a : {delta_min:.0f} minutes
- Calibration : {calib}
## 📍 NEWS INTER-CYCLE — dernières {delta_min:.0f}min (POTENTIELLEMENT PAS ENCORE dans les prix)
{_fmt_news(inter, 6)}
*→ Ce sont les nouvelles les plus susceptibles de créer des opportunités non-pricées.*
## 📰 NEWS RÉCENTES — 24h (partiellement pricées, decay appliqué)
{_fmt_news(recent, 5)}
## 📦 NEWS PLUS ANCIENNES — >24h (considérées comme déjà intégrées par le marché)
{_fmt_news(older, 3)}
"""
return block
def suggest_patterns_from_market_context(
news: List[Dict],
quotes_by_class: Dict[str, List[Dict]],
@@ -819,11 +963,23 @@ def suggest_patterns_from_market_context(
portfolio_lessons: Optional[Dict] = None,
reliability_map: Optional[Dict] = None,
iv_context: str = "",
cycle_meta: Optional[Dict] = None,
) -> List[Dict]:
"""Ask GPT-4o to propose new patterns based on current geo/market + macro regime context."""
top_news = sorted(news, key=lambda x: x.get("impact_score", 0), reverse=True)[:12]
_cycle_meta = cycle_meta or {}
_delta_min = _cycle_meta.get("delta_minutes", 180.0)
# Apply temporal decay + partition news by age relative to cycle delta
news_decayed = apply_news_decay(news)
_partitioned = partition_news_by_age(news_decayed, _delta_min)
# For backward-compat blocks that still use a flat list: use decayed inter+recent
top_news = sorted(
_partitioned["inter_cycle"] + _partitioned["recent_24h"],
key=lambda x: -(x.get("decayed_score") or 0)
)[:12]
news_block = "\n".join([
f"- [{n.get('source','')}] {n.get('title','')} (impact {n.get('impact_score',0):.2f})"
f"- [{n.get('source','')}] {n.get('title','')} (impact_decay {n.get('decayed_score',0):.2f})"
for n in top_news
])
@@ -948,11 +1104,14 @@ Règles supplémentaires:
⚠️ INTERDICTION: Ne jamais suggérer "Long Call" ou "Long Put" (naked) si IVR > 60% sur ce ticker.
"""
# Build temporal news block (partitioned by cycle delta)
temporal_news_block = _build_temporal_news_block(_partitioned, _cycle_meta) if _cycle_meta else (
"## Actualités géopolitiques du moment (triées par impact)\n" + news_block
)
user = f"""Tu es un stratège géopolitique et financier senior, expert en options.
{macro_block}{geo_block}{lessons_block}{reliability_block}{iv_block}
## Actualités géopolitiques du moment (triées par impact)
{news_block}
{temporal_news_block}
## Prix des marchés (variation J-1)
{market_block}

37
backend/services/auto_cycle.py
View File

@@ -174,6 +174,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
save_pattern_scores, log_macro_regime, log_geo_alert, log_trade_entries,
add_cycle_run, update_cycle_run, save_reasoning_trace,
get_latest_portfolio_lessons, log_system_event,
get_last_completed_cycle_ts,
)
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
@@ -206,6 +207,37 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
_current_status["running"] = True
_current_status["last_run_id"] = run_id
# ── Cycle meta — timing context ───────────────────────────────────────
_now = datetime.utcnow()
_last_cycle_ts_str = get_last_completed_cycle_ts()
_delta_minutes: float = 180.0 # default 3h if no prior cycle
if _last_cycle_ts_str:
try:
_last_dt = datetime.fromisoformat(_last_cycle_ts_str.replace("Z", ""))
_delta_minutes = max(1.0, (_now - _last_dt).total_seconds() / 60)
except Exception:
pass
_interval_hours = float(get_config("auto_cycle_interval_hours") or "3")
if _delta_minutes < 90:
_calib_label = "Court terme (<2h) — signaux très récents"
elif _delta_minutes < 720:
_calib_label = f"Moyen terme ({_delta_minutes/60:.0f}h) — signaux potentiellement pas encore pricés"
else:
_calib_label = f"Long terme ({_delta_minutes/60:.0f}h) — marchés ont eu le temps d'intégrer"
cycle_meta = {
"current_cycle_ts": _now.isoformat(),
"last_cycle_ts": _last_cycle_ts_str,
"delta_minutes": round(_delta_minutes, 1),
"interval_hours": _interval_hours,
"calibration_label": _calib_label,
}
logger.info(
f"[Cycle {run_id[:16]}] Cycle meta: delta={_delta_minutes:.0f}min depuis dernier cycle"
f" ({_last_cycle_ts_str[:16] if _last_cycle_ts_str else 'premier cycle'})"
)
# ── Step 0: Load portfolio lessons + Super Contexte ──────────────────
portfolio_lessons = get_latest_portfolio_lessons()
@@ -341,11 +373,15 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
try:
from services.data_fetcher import get_economic_calendar
calendar = get_economic_calendar()
# Apply decay to news before suggestion (adds decayed_score + age_hours)
from services.ai_analyzer import apply_news_decay as _apply_decay
news = _apply_decay(news)
suggestions = suggest_patterns_from_market_context(
news, quotes, calendar, macro_regime=macro_regime, geo_score=geo_score_obj,
portfolio_lessons=portfolio_lessons,
reliability_map=_reliability_map or None,
iv_context=iv_context,
cycle_meta=cycle_meta,
)
except Exception as e:
logger.warning(f"[Cycle] Suggestion step failed: {e}")
@@ -468,6 +504,7 @@ def run_cycle_once(trigger: str = "auto") -> Dict[str, Any]:
portfolio_lessons=portfolio_lessons,
iv_context=iv_context,
risk_context=risk_cluster_context,
cycle_meta=cycle_meta,
)
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")]

10
backend/services/database.py
View File

@@ -1212,6 +1212,16 @@ def reset_journal_history():
conn.close()
def get_last_completed_cycle_ts() -> Optional[str]:
"""Return ISO timestamp of the last successfully completed cycle, or None."""
conn = get_conn()
row = conn.execute(
"SELECT completed_at FROM cycle_runs WHERE status='completed' ORDER BY completed_at DESC LIMIT 1"
).fetchone()
conn.close()
return row["completed_at"] if row else None
def add_cycle_run(run_id: str, trigger: str = "auto") -> None:
conn = get_conn()
conn.execute(