89 lines
3.1 KiB
Python
89 lines
3.1 KiB
Python
from fastapi import APIRouter, Query
|
|
from typing import Optional, List
|
|
from services.data_fetcher import fetch_geo_news, get_economic_calendar
|
|
from services.geo_analyzer import compute_geo_risk_score, match_patterns, generate_trade_ideas, compute_pattern_relevance
|
|
from services.database import get_custom_patterns
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
router = APIRouter(prefix="/api/geo", tags=["geopolitical"])
|
|
|
|
_news_cache: dict = {"data": [], "ts": 0}
|
|
|
|
|
|
@router.get("/news")
|
|
def geo_news(force_refresh: bool = False):
|
|
import time
|
|
now = time.time()
|
|
if not force_refresh and _news_cache["data"] and (now - _news_cache["ts"]) < 3600:
|
|
return _news_cache["data"]
|
|
news = fetch_geo_news()
|
|
# Enrich with AI: corrects impact_score, adds ai_dir_energy/metals/indices,
|
|
# ai_resolution (ceasefire/peace deal), ai_insight (1 French sentence).
|
|
# Gracefully skips if OpenAI not configured.
|
|
try:
|
|
from services.ai_analyzer import ai_score_news_batch
|
|
news = ai_score_news_batch(news)
|
|
except Exception:
|
|
pass
|
|
_news_cache["data"] = news
|
|
_news_cache["ts"] = now
|
|
return news
|
|
|
|
|
|
@router.get("/risk-score")
|
|
def risk_score():
|
|
"""Frozen per-cycle AI-judged score (services.ai_analyzer.ai_score_geo_risk,
|
|
saved once per auto-cycle run in geo_risk_snapshots) — no longer
|
|
recomputed live on every request. Falls back to a live algorithmic
|
|
compute only if no cycle has ever run yet (e.g. fresh install)."""
|
|
from services.database import get_latest_geo_risk_snapshot
|
|
snapshot = get_latest_geo_risk_snapshot()
|
|
if snapshot:
|
|
return snapshot
|
|
news = _news_cache["data"] or fetch_geo_news()
|
|
return compute_geo_risk_score(news)
|
|
|
|
|
|
@router.get("/pattern-matches")
|
|
def pattern_matches():
|
|
news = _news_cache["data"] or fetch_geo_news()
|
|
all_patterns = get_custom_patterns()
|
|
return match_patterns(news, patterns=all_patterns)
|
|
|
|
|
|
@router.get("/pattern-relevance")
|
|
def pattern_relevance(days: int = 2):
|
|
"""Return ALL active patterns with news-keyword relevance over the last N days."""
|
|
all_news = _news_cache["data"] or fetch_geo_news()
|
|
# Filter news to last N days
|
|
if days > 0:
|
|
cutoff = datetime.now(timezone.utc) - timedelta(days=days)
|
|
recent: list = []
|
|
for n in all_news:
|
|
try:
|
|
from email.utils import parsedate_to_datetime
|
|
d = parsedate_to_datetime(str(n.get("date", "")))
|
|
if d >= cutoff:
|
|
recent.append(n)
|
|
except Exception:
|
|
recent.append(n)
|
|
news = recent if recent else all_news
|
|
else:
|
|
news = all_news
|
|
all_patterns = get_custom_patterns()
|
|
return compute_pattern_relevance(news, patterns=all_patterns)
|
|
|
|
|
|
@router.get("/trade-ideas")
|
|
def trade_ideas():
|
|
news = _news_cache["data"] or fetch_geo_news()
|
|
all_patterns = get_custom_patterns()
|
|
matches = match_patterns(news, patterns=all_patterns)
|
|
geo_score = compute_geo_risk_score(news)
|
|
return generate_trade_ideas(matches, geo_score)
|
|
|
|
|
|
@router.get("/calendar")
|
|
def calendar():
|
|
return get_economic_calendar()
|