d256b65d30
Stack: FastAPI + React/TypeScript + SQLite + GPT-4o Features: Radar géopolitique, Marchés, Régime Macro, Journal de Bord MTM, Rapport IA, Super Contexte (base de raisonnement évolutive), Boucle feedback IA. Deploy: Docker + docker-compose + nginx pour openfin.open-squared.tech Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
349 lines
16 KiB
Python
349 lines
16 KiB
Python
"""
|
|
Geopolitical pattern engine.
|
|
Scores current events against historical templates and generates trade signals.
|
|
"""
|
|
from datetime import datetime, timedelta
|
|
from typing import List, Dict, Any, Optional
|
|
import json
|
|
|
|
|
|
# ── Historical geopolitical pattern library ───────────────────────────────────
|
|
GEO_PATTERNS = [
|
|
{
|
|
"id": "P001",
|
|
"name": "Middle East Military Escalation → Oil Spike",
|
|
"description": "Armed conflict or threat in Gulf region triggers Brent/WTI crude spike +10-20% within 2-4 weeks",
|
|
"triggers": ["military", "energy", "sanctions"],
|
|
"keywords": ["Iran", "Israel", "Saudi", "Gulf", "Strait of Hormuz", "OPEC"],
|
|
"historical_instances": [
|
|
{"date": "2019-09-14", "event": "Attack on Saudi Aramco facilities", "brent_move": +14.6, "days": 2},
|
|
{"date": "2020-01-03", "event": "Soleimani assassination", "brent_move": +4.4, "days": 1},
|
|
{"date": "2022-02-24", "event": "Russia invades Ukraine", "brent_move": +28.0, "days": 10},
|
|
],
|
|
"suggested_trades": [
|
|
{"strategy": "Bull Call Spread", "underlying": "USO", "rationale": "Oil ETF call spread, limited risk"},
|
|
{"strategy": "Long Call", "underlying": "CL=F", "rationale": "WTI crude direct exposure"},
|
|
],
|
|
"asset_class": "energy",
|
|
"expected_move_pct": 12.0,
|
|
"probability": 0.65,
|
|
"horizon_days": 30,
|
|
},
|
|
{
|
|
"id": "P002",
|
|
"name": "US Tariff Announcement → Agriculture Selloff",
|
|
"description": "Trump/US tariff threats on China cause immediate selloff in soy, corn, wheat (retaliatory risk)",
|
|
"triggers": ["trade_war", "political_speech"],
|
|
"keywords": ["tariff", "China", "trade", "soybean", "agriculture", "import duty"],
|
|
"historical_instances": [
|
|
{"date": "2018-07-06", "event": "US-China trade war tariffs", "zs_move": -10.2, "days": 30},
|
|
{"date": "2019-05-10", "event": "Trump tariff escalation tweet", "zs_move": -5.8, "days": 5},
|
|
{"date": "2025-02-01", "event": "Trump 25% tariff on Canada/Mexico", "zw_move": -3.4, "days": 3},
|
|
],
|
|
"suggested_trades": [
|
|
{"strategy": "Bear Put Spread", "underlying": "SOYB", "rationale": "Downside hedge on soy ETF"},
|
|
{"strategy": "Long Put", "underlying": "ZS=F", "rationale": "Soybean futures put"},
|
|
],
|
|
"asset_class": "agriculture",
|
|
"expected_move_pct": -8.0,
|
|
"probability": 0.70,
|
|
"horizon_days": 21,
|
|
},
|
|
{
|
|
"id": "P003",
|
|
"name": "Geopolitical Risk Flight → Gold Rally",
|
|
"description": "Major geopolitical uncertainty drives safe-haven demand for gold +5-15%",
|
|
"triggers": ["military", "health_crisis", "financial_crisis", "elections"],
|
|
"keywords": ["nuclear", "war", "crisis", "uncertainty", "safe haven", "debt ceiling"],
|
|
"historical_instances": [
|
|
{"date": "2022-02-24", "event": "Ukraine invasion", "gc_move": +6.8, "days": 14},
|
|
{"date": "2023-10-07", "event": "Hamas attack on Israel", "gc_move": +9.2, "days": 30},
|
|
{"date": "2020-03-01", "event": "COVID-19 fear peak", "gc_move": +12.1, "days": 45},
|
|
],
|
|
"suggested_trades": [
|
|
{"strategy": "Long Call", "underlying": "GLD", "rationale": "Gold ETF call for safe-haven rally"},
|
|
{"strategy": "Bull Call Spread", "underlying": "GC=F", "rationale": "Gold futures spread, capped risk"},
|
|
],
|
|
"asset_class": "metals",
|
|
"expected_move_pct": 7.5,
|
|
"probability": 0.72,
|
|
"horizon_days": 30,
|
|
},
|
|
{
|
|
"id": "P004",
|
|
"name": "Fed Hawkish Pivot → Dollar Surge / EM Currency Crash",
|
|
"description": "Fed signals higher-for-longer rates → USD Index rallies, EUR/USD drops",
|
|
"triggers": ["political_speech"],
|
|
"keywords": ["Fed", "interest rate", "hike", "hawkish", "inflation", "FOMC", "Powell"],
|
|
"historical_instances": [
|
|
{"date": "2022-06-15", "event": "Fed 75bps hike", "dxy_move": +3.2, "days": 5},
|
|
{"date": "2023-03-22", "event": "Fed signals further hikes", "eurusd_move": -1.8, "days": 7},
|
|
],
|
|
"suggested_trades": [
|
|
{"strategy": "Bear Put Spread", "underlying": "FXE", "rationale": "EUR/USD put spread"},
|
|
{"strategy": "Long Call", "underlying": "UUP", "rationale": "Dollar index ETF call"},
|
|
],
|
|
"asset_class": "forex",
|
|
"expected_move_pct": 3.0,
|
|
"probability": 0.68,
|
|
"horizon_days": 14,
|
|
},
|
|
{
|
|
"id": "P005",
|
|
"name": "China Economic Slowdown → Copper/Metals Selloff",
|
|
"description": "Weak Chinese PMI or stimulus disappointment drives copper lower (China = 50%+ of global demand)",
|
|
"triggers": ["resource_scarcity", "trade_war"],
|
|
"keywords": ["China", "PMI", "slowdown", "recession", "property", "Evergrande", "copper demand"],
|
|
"historical_instances": [
|
|
{"date": "2015-08-24", "event": "China Black Monday", "hg_move": -8.4, "days": 5},
|
|
{"date": "2022-11-01", "event": "China PMI contraction", "hg_move": -5.2, "days": 10},
|
|
],
|
|
"suggested_trades": [
|
|
{"strategy": "Long Put", "underlying": "COPX", "rationale": "Copper miners ETF put"},
|
|
{"strategy": "Bear Put Spread", "underlying": "HG=F", "rationale": "Copper futures spread"},
|
|
],
|
|
"asset_class": "metals",
|
|
"expected_move_pct": -6.5,
|
|
"probability": 0.60,
|
|
"horizon_days": 21,
|
|
},
|
|
{
|
|
"id": "P006",
|
|
"name": "Ukraine/Russia War Escalation → Wheat Spike + Defense Rally",
|
|
"description": "New escalation in Russia-Ukraine conflict → wheat/fertilizer spike, defense stocks rally",
|
|
"triggers": ["military", "resource_scarcity"],
|
|
"keywords": ["Russia", "Ukraine", "Zelensky", "Kyiv", "grain corridor", "Black Sea", "NATO"],
|
|
"historical_instances": [
|
|
{"date": "2022-02-24", "event": "Full-scale invasion", "zw_move": +50.0, "days": 45},
|
|
{"date": "2022-07-22", "event": "Grain deal collapse threat", "zw_move": +6.3, "days": 3},
|
|
{"date": "2023-07-17", "event": "Russia exits grain deal", "zw_move": +8.5, "days": 2},
|
|
],
|
|
"suggested_trades": [
|
|
{"strategy": "Long Call", "underlying": "WEAT", "rationale": "Wheat ETF call on supply shock"},
|
|
{"strategy": "Bull Call Spread", "underlying": "LMT", "rationale": "Lockheed defense stock spread"},
|
|
],
|
|
"asset_class": "agriculture",
|
|
"expected_move_pct": 15.0,
|
|
"probability": 0.58,
|
|
"horizon_days": 45,
|
|
},
|
|
{
|
|
"id": "P007",
|
|
"name": "Natural Gas Supply Disruption → NG Price Spike",
|
|
"description": "Pipeline disruption, LNG strike, or extreme weather drives natural gas +20-40%",
|
|
"triggers": ["energy", "natural_disaster", "military"],
|
|
"keywords": ["pipeline", "LNG", "natural gas", "Nord Stream", "gas supply", "storage"],
|
|
"historical_instances": [
|
|
{"date": "2022-09-26", "event": "Nord Stream pipeline explosion", "ng_move": +18.0, "days": 5},
|
|
{"date": "2021-02-10", "event": "Texas winter storm Uri", "ng_move": +40.0, "days": 3},
|
|
],
|
|
"suggested_trades": [
|
|
{"strategy": "Long Call", "underlying": "UNG", "rationale": "Natural gas ETF call"},
|
|
{"strategy": "Bull Call Spread", "underlying": "NG=F", "rationale": "NG futures spread, capped risk"},
|
|
],
|
|
"asset_class": "energy",
|
|
"expected_move_pct": 25.0,
|
|
"probability": 0.55,
|
|
"horizon_days": 14,
|
|
},
|
|
{
|
|
"id": "P008",
|
|
"name": "Pandemic / Health Crisis → VIX Spike + Market Selloff",
|
|
"description": "New pandemic scare or major health crisis → VIX spike, equity selloff, gold bid",
|
|
"triggers": ["health_crisis"],
|
|
"keywords": ["pandemic", "virus", "outbreak", "WHO", "lockdown", "COVID", "mpox", "H5N1"],
|
|
"historical_instances": [
|
|
{"date": "2020-02-24", "event": "COVID-19 global spread fear", "spx_move": -34.0, "days": 30},
|
|
{"date": "2022-11-25", "event": "China COVID lockdowns", "spx_move": -3.5, "days": 3},
|
|
],
|
|
"suggested_trades": [
|
|
{"strategy": "Long Put", "underlying": "SPY", "rationale": "S&P 500 put for equity protection"},
|
|
{"strategy": "Long Call", "underlying": "^VIX", "rationale": "VIX call for volatility spike"},
|
|
{"strategy": "Long Call", "underlying": "GLD", "rationale": "Gold safe-haven call"},
|
|
],
|
|
"asset_class": "indices",
|
|
"expected_move_pct": -12.0,
|
|
"probability": 0.45,
|
|
"horizon_days": 30,
|
|
},
|
|
]
|
|
|
|
|
|
GEOPOLITICAL_RISK_WEIGHTS = {
|
|
"military": 0.25,
|
|
"energy": 0.20,
|
|
"trade_war": 0.15,
|
|
"political_speech": 0.15,
|
|
"natural_disaster": 0.10,
|
|
"health_crisis": 0.10,
|
|
"resource_scarcity": 0.05,
|
|
}
|
|
|
|
|
|
def compute_geo_risk_score(events: List[Dict[str, Any]]) -> Dict[str, Any]:
|
|
"""Compute a global geopolitical risk score 0-100 from recent events."""
|
|
if not events:
|
|
return {"score": 35, "level": "medium", "breakdown": {}}
|
|
|
|
category_scores: Dict[str, float] = {}
|
|
for event in events[:30]:
|
|
cat = event.get("category", "general")
|
|
impact = event.get("impact_score", 0.1)
|
|
if cat in category_scores:
|
|
category_scores[cat] = max(category_scores[cat], impact)
|
|
else:
|
|
category_scores[cat] = impact
|
|
|
|
weighted = sum(
|
|
category_scores.get(cat, 0) * weight
|
|
for cat, weight in GEOPOLITICAL_RISK_WEIGHTS.items()
|
|
)
|
|
score = min(100, round(weighted * 100, 1))
|
|
|
|
if score < 25:
|
|
level = "low"
|
|
elif score < 50:
|
|
level = "medium"
|
|
elif score < 75:
|
|
level = "high"
|
|
else:
|
|
level = "extreme"
|
|
|
|
return {
|
|
"score": score,
|
|
"level": level,
|
|
"breakdown": {cat: round(v * 100, 1) for cat, v in category_scores.items()},
|
|
"top_risks": sorted(category_scores.items(), key=lambda x: x[1], reverse=True)[:3],
|
|
}
|
|
|
|
|
|
def match_patterns(events: List[Dict[str, Any]], patterns: Optional[List[Dict[str, Any]]] = None) -> List[Dict[str, Any]]:
|
|
"""Find which historical geo-patterns best match current event feed."""
|
|
if not events:
|
|
return []
|
|
if patterns is None:
|
|
patterns = GEO_PATTERNS
|
|
|
|
current_categories = set(e.get("category", "") for e in events)
|
|
current_tags = set()
|
|
for e in events:
|
|
current_tags.update(e.get("tags", []))
|
|
current_text = " ".join(e.get("title", "") + " " + e.get("summary", "") for e in events[:20]).lower()
|
|
|
|
matches = []
|
|
for pattern in patterns:
|
|
trigger_match = len(set(pattern["triggers"]) & current_categories) / len(pattern["triggers"])
|
|
keyword_match = sum(1 for kw in pattern["keywords"] if kw.lower() in current_text) / len(pattern["keywords"])
|
|
similarity = round((trigger_match * 0.5 + keyword_match * 0.5) * 100, 1)
|
|
|
|
if similarity > 10:
|
|
matches.append({
|
|
"pattern_id": pattern["id"],
|
|
"name": pattern["name"],
|
|
"description": pattern["description"],
|
|
"similarity": similarity,
|
|
"suggested_trades": pattern["suggested_trades"],
|
|
"asset_class": pattern["asset_class"],
|
|
"expected_move_pct": pattern["expected_move_pct"],
|
|
"probability": pattern["probability"],
|
|
"horizon_days": pattern["horizon_days"],
|
|
"historical_instances": pattern["historical_instances"],
|
|
})
|
|
|
|
return sorted(matches, key=lambda x: x["similarity"], reverse=True)[:5]
|
|
|
|
|
|
def generate_trade_ideas(pattern_matches: List[Dict[str, Any]], geo_score: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|
"""Convert pattern matches into structured trade ideas with sizing for ~1000€."""
|
|
ideas = []
|
|
for pm in pattern_matches[:5]:
|
|
for i, trade in enumerate(pm["suggested_trades"]): # all suggested trades, not just first
|
|
move = pm["expected_move_pct"]
|
|
confidence = round(pm["probability"] * pm["similarity"] / 100 * 100)
|
|
# Use trade-level asset_class if provided, else fall back to pattern-level
|
|
asset_class = trade.get("asset_class") or pm["asset_class"]
|
|
ideas.append({
|
|
"id": f"IDEA-{pm['pattern_id']}-{i}-{trade['strategy'][:3].upper()}",
|
|
"title": f"{trade['strategy']} on {trade['underlying']}",
|
|
"rationale": f"[{pm['name']}] {trade['rationale']}. Expected move: {'+' if move > 0 else ''}{move}% in {pm['horizon_days']}d",
|
|
"pattern": pm["name"],
|
|
"asset_class": asset_class,
|
|
"underlying": trade["underlying"],
|
|
"strategy": trade["strategy"],
|
|
"expected_move_pct": move,
|
|
"confidence": min(95, confidence),
|
|
"horizon_days": pm["horizon_days"],
|
|
"capital_required": 1000,
|
|
"risk_level": "high" if abs(move) > 15 else "medium",
|
|
"pattern_similarity": pm["similarity"],
|
|
})
|
|
return ideas
|
|
|
|
|
|
def compute_pattern_relevance(
|
|
events: List[Dict[str, Any]],
|
|
patterns: Optional[List[Dict[str, Any]]] = None,
|
|
) -> List[Dict[str, Any]]:
|
|
"""Return ALL patterns with news-keyword relevance score + matching news snippets.
|
|
Unlike match_patterns(), no similarity threshold — every active pattern is returned.
|
|
"""
|
|
if patterns is None:
|
|
patterns = GEO_PATTERNS
|
|
|
|
current_categories = set(e.get("category", "") for e in events)
|
|
current_text = " ".join(
|
|
e.get("title", "") + " " + e.get("summary", "") for e in events[:30]
|
|
).lower()
|
|
|
|
result = []
|
|
for pattern in patterns:
|
|
triggers_list = pattern.get("triggers", []) or []
|
|
keywords_list = pattern.get("keywords", []) or []
|
|
|
|
trigger_match = (
|
|
len(set(triggers_list) & current_categories) / len(triggers_list)
|
|
if triggers_list else 0
|
|
)
|
|
kw_hits = [kw for kw in keywords_list if kw.lower() in current_text]
|
|
keyword_match = len(kw_hits) / len(keywords_list) if keywords_list else 0
|
|
relevance = round((trigger_match * 0.5 + keyword_match * 0.5) * 100, 1)
|
|
|
|
# Find matching news with which keywords triggered
|
|
matching_news = []
|
|
for e in events[:30]:
|
|
text = (e.get("title", "") + " " + e.get("summary", "")).lower()
|
|
hits = [kw for kw in keywords_list if kw.lower() in text]
|
|
if hits:
|
|
matching_news.append({
|
|
"title": e.get("title", ""),
|
|
"source": e.get("source", ""),
|
|
"date": str(e.get("date", ""))[:16],
|
|
"impact": round(e.get("impact_score", 0), 2),
|
|
"matched_keywords": hits,
|
|
"url": e.get("url", ""),
|
|
})
|
|
matching_news.sort(key=lambda x: x["impact"], reverse=True)
|
|
|
|
result.append({
|
|
"pattern_id": pattern.get("id", ""),
|
|
"name": pattern.get("name", ""),
|
|
"description": pattern.get("description", ""),
|
|
"asset_class": pattern.get("asset_class", ""),
|
|
"relevance": relevance,
|
|
"keyword_hits": len(kw_hits),
|
|
"keyword_total": len(keywords_list),
|
|
"matched_keywords": kw_hits,
|
|
"matching_news": matching_news[:5],
|
|
"suggested_trades": pattern.get("suggested_trades", []),
|
|
"expected_move_pct": pattern.get("expected_move_pct", 0),
|
|
"probability": pattern.get("probability", 0),
|
|
"horizon_days": pattern.get("horizon_days", 0),
|
|
})
|
|
|
|
result.sort(key=lambda x: x["relevance"], reverse=True)
|
|
return result
|
|
|
|
|
|
def get_all_patterns() -> List[Dict[str, Any]]:
|
|
return GEO_PATTERNS
|