""" Phase 4 — Price Discovery Status For each scored news article, we track the prices of related tickers at capture time. On subsequent cycles we measure how much of the expected move has already happened ("absorption"), and flag opportunities where price has NOT yet moved. Direction mapping from ai_score_news_batch fields: ai_dir_energy → BZ=F (Brent), NG=F (Natural Gas) ai_dir_metals → GC=F (Gold), HG=F (Copper) ai_dir_indices → ^GSPC (S&P 500), IWM (Russell 2000) ai_dir_forex → DX-Y.NYB (DXY) [if present] """ from __future__ import annotations import hashlib import logging import math from datetime import datetime, timezone from typing import Any, Dict, List, Optional logger = logging.getLogger("price_discovery") # Map AI direction fields → related tickers DIR_FIELD_TO_TICKERS: Dict[str, list] = { "ai_dir_energy": ["BZ=F", "NG=F"], "ai_dir_metals": ["GC=F", "HG=F"], "ai_dir_indices": ["^GSPC", "IWM"], } # Minimum impact score to bother tracking MIN_IMPACT_SCORE = 0.4 # Absorption thresholds FULLY_PRICED_THRESHOLD = 0.80 # >80% absorbed PARTIALLY_PRICED = 0.30 # 30-80% # <30% → not_yet_priced # Expected move % per asset class per unit of impact score # e.g. energy with impact 0.8 → expected ~2.0% move EXPECTED_MOVE_PCT: Dict[str, float] = { "BZ=F": 2.5, "NG=F": 3.5, "GC=F": 1.2, "HG=F": 2.0, "^GSPC": 1.5, "IWM": 2.0, "DX-Y.NYB": 0.8, } def _article_hash(article: Dict) -> str: key = (article.get("title", "") + article.get("source", "")).encode() return hashlib.md5(key).hexdigest()[:16] def capture_price_snapshots(news: List[Dict], quotes_flat: Dict[str, float], cycle_id: str) -> int: """ For each scored news article, save a price snapshot for related tickers. quotes_flat: {ticker_symbol: price} — pre-flattened from get_all_quotes() Returns number of snapshots saved. """ from services.database import save_news_price_snapshot saved = 0 for article in news: impact = float(article.get("impact_score") or 0) if impact < MIN_IMPACT_SCORE: continue ah = _article_hash(article) title = article.get("title", "")[:200] for dir_field, tickers in DIR_FIELD_TO_TICKERS.items(): direction = article.get(dir_field, "neutral") if direction == "neutral": continue for ticker in tickers: price = quotes_flat.get(ticker) if price is None: continue try: save_news_price_snapshot( article_hash=ah, article_title=title, ticker=ticker, expected_direction=direction, expected_impact_score=round(impact, 3), price_at_capture=price, cycle_id=cycle_id, ) saved += 1 except Exception as e: logger.debug(f"[PD] snapshot save failed {ticker}: {e}") return saved def _fetch_current_prices(tickers: List[str]) -> Dict[str, float]: """Fetch latest prices via yfinance for a list of tickers.""" if not tickers: return {} try: import yfinance as yf data = yf.download(tickers, period="2d", interval="1d", progress=False, auto_adjust=True) prices: Dict[str, float] = {} if hasattr(data.columns, "levels"): # MultiIndex: (field, ticker) close = data["Close"] if "Close" in data else data for tkr in tickers: try: val = float(close[tkr].dropna().iloc[-1]) prices[tkr] = val except Exception: pass else: # Single ticker try: val = float(data["Close"].dropna().iloc[-1]) prices[tickers[0]] = val except Exception: pass return prices except Exception as e: logger.warning(f"[PD] yfinance price fetch failed: {e}") return {} def compute_absorptions(min_age_minutes: float = 30.0, max_age_days: int = 7) -> List[Dict]: """ For all snapshots in the correct age window, compute absorption. Returns list of dicts with absorption data, sorted by opportunity (lowest absorption first). """ from services.database import get_news_price_snapshots snapshots = get_news_price_snapshots(max_age_days=max_age_days, min_age_minutes=min_age_minutes) if not snapshots: return [] tickers = list({s["ticker"] for s in snapshots}) current_prices = _fetch_current_prices(tickers) results = [] for snap in snapshots: ticker = snap["ticker"] price_now = current_prices.get(ticker) if price_now is None: continue price_cap = snap["price_at_capture"] if not price_cap or price_cap == 0: continue direction = snap["expected_direction"] # bullish | bearish impact = snap["expected_impact_score"] expected_move = EXPECTED_MOVE_PCT.get(ticker, 1.5) * impact actual_move_pct = (price_now - price_cap) / price_cap * 100 # Align actual move with expected direction signed_move = actual_move_pct if direction == "bullish" else -actual_move_pct absorption = max(0.0, signed_move / expected_move) if expected_move > 0 else 0.0 if absorption >= FULLY_PRICED_THRESHOLD: status = "fully_priced" elif absorption >= PARTIALLY_PRICED: status = "partially_priced" else: status = "not_yet_priced" results.append({ "ticker": ticker, "article_title": snap["article_title"], "article_hash": snap["article_hash"], "expected_direction": direction, "expected_impact_score": impact, "price_at_capture": round(price_cap, 4), "price_now": round(price_now, 4), "actual_move_pct": round(actual_move_pct, 3), "signed_move_pct": round(signed_move, 3), "expected_move_pct": round(expected_move, 3), "absorption_pct": round(min(absorption * 100, 200), 1), "status": status, "opportunity": status == "not_yet_priced", "captured_at": snap["captured_at"], }) # Sort: opportunities first (lowest absorption), then partially priced results.sort(key=lambda r: r["absorption_pct"]) return results def build_price_discovery_block(absorptions: List[Dict]) -> str: """Build the prompt block from computed absorptions.""" if not absorptions: return "" opportunities = [a for a in absorptions if a["status"] == "not_yet_priced"] partial = [a for a in absorptions if a["status"] == "partially_priced"] priced = [a for a in absorptions if a["status"] == "fully_priced"] lines = ["## ⚡ PRICE DISCOVERY STATUS — Signaux déjà dans les prix ou pas ?"] if opportunities: lines.append(f"\n🔥 NON ENCORE PRICÉS ({len(opportunities)}) — OPPORTUNITÉS POTENTIELLES :") for a in opportunities[:5]: lines.append( f" {a['ticker']} | \"{a['article_title'][:55]}...\" " f"→ dir {a['expected_direction'].upper()} | mouvement actuel {a['actual_move_pct']:+.2f}% " f"vs attendu {a['expected_move_pct']:+.2f}% → {a['absorption_pct']:.0f}% absorbé" ) lines.append(" ⚠️ Ces tickers ont reçu un signal fort mais le marché n'a pas encore bougé.") if partial: lines.append(f"\n⚠️ PARTIELLEMENT PRICÉS ({len(partial)}) — FENÊTRE EN COURS :") for a in partial[:3]: lines.append( f" {a['ticker']} → {a['absorption_pct']:.0f}% absorbé ({a['actual_move_pct']:+.2f}% / {a['expected_move_pct']:+.2f}% attendu)" ) if priced: lines.append(f"\n✅ DÉJÀ PRICÉS ({len(priced)}) — éviter de chasser :") for a in priced[:3]: lines.append( f" {a['ticker']} → {a['absorption_pct']:.0f}% absorbé — marché a déjà intégré" ) lines.append( "\n⚠️ CONSIGNE : Pour les tickers NON ENCORE PRICÉS, tu peux être plus agressif sur l'expected_move_pct. " "Pour les tickers DÉJÀ PRICÉS, évite les positions dans le sens de la news (momentum tardif)." ) return "\n".join(lines)