feat: bank forceasts
This commit is contained in:
OpenSquared
@@ -235,3 +235,127 @@ def score_text(body: ScoreTextRequest):
|
||||
except Exception as e:
|
||||
logger.error(f"Text scoring error: {e}")
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
|
||||
# ── Bank Forecasts ─────────────────────────────────────────────────────────────
|
||||
|
||||
class BankSourceUpsert(BaseModel):
|
||||
name: str
|
||||
url: str = ""
|
||||
active: bool = True
|
||||
notes: str = ""
|
||||
|
||||
|
||||
@router.get("/bank-forecasts/sources")
|
||||
def list_bank_sources():
|
||||
from services.database import get_conn
|
||||
conn = get_conn()
|
||||
try:
|
||||
rows = conn.execute(
|
||||
"SELECT id, name, url, active, last_scraped, notes FROM bank_forecast_sources ORDER BY name"
|
||||
).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
@router.put("/bank-forecasts/sources/{source_id}")
|
||||
def upsert_bank_source(source_id: str, body: BankSourceUpsert):
|
||||
from services.database import get_conn
|
||||
conn = get_conn()
|
||||
try:
|
||||
conn.execute(
|
||||
"""INSERT INTO bank_forecast_sources (id, name, url, active, notes)
|
||||
VALUES (?,?,?,?,?)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
name=excluded.name, url=excluded.url,
|
||||
active=excluded.active, notes=excluded.notes""",
|
||||
(source_id, body.name, body.url, int(body.active), body.notes)
|
||||
)
|
||||
conn.commit()
|
||||
return {"saved": source_id}
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
@router.get("/bank-forecasts")
|
||||
def list_bank_forecasts(series_id: str = "", event_date: str = ""):
|
||||
from services.database import get_conn
|
||||
conn = get_conn()
|
||||
try:
|
||||
wheres = ["1=1"]
|
||||
params = []
|
||||
if series_id:
|
||||
wheres.append("bf.series_id = ?"); params.append(series_id)
|
||||
if event_date:
|
||||
wheres.append("bf.event_date = ?"); params.append(event_date)
|
||||
rows = conn.execute(
|
||||
f"""SELECT bf.*, bs.name as bank_name
|
||||
FROM bank_forecasts bf
|
||||
JOIN bank_forecast_sources bs ON bs.id = bf.source_id
|
||||
WHERE {' AND '.join(wheres)}
|
||||
ORDER BY bf.event_date DESC, bf.extracted_at DESC
|
||||
LIMIT 500""",
|
||||
params
|
||||
).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
@router.get("/bank-forecasts/consensus")
|
||||
def get_consensus():
|
||||
"""Return one consensus row per (series_id, event_date) = average of bank forecasts."""
|
||||
from services.database import get_conn
|
||||
conn = get_conn()
|
||||
try:
|
||||
rows = conn.execute(
|
||||
"""SELECT series_id, event_name, event_date,
|
||||
ROUND(AVG(forecast_value), 4) as consensus,
|
||||
COUNT(*) as bank_count,
|
||||
MIN(forecast_value) as min_forecast,
|
||||
MAX(forecast_value) as max_forecast
|
||||
FROM bank_forecasts
|
||||
WHERE forecast_value IS NOT NULL
|
||||
GROUP BY series_id, event_date
|
||||
ORDER BY event_date DESC"""
|
||||
).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
@router.post("/bank-forecasts/scrape")
|
||||
def trigger_scrape(source_id: str = ""):
|
||||
"""Scrape one source (source_id) or all active sources (empty)."""
|
||||
from services.database import get_conn
|
||||
from services.bank_forecast_scraper import scrape_source, scrape_all_active
|
||||
conn = get_conn()
|
||||
try:
|
||||
if source_id:
|
||||
s = conn.execute("SELECT * FROM bank_forecast_sources WHERE id=?", (source_id,)).fetchone()
|
||||
if not s:
|
||||
raise HTTPException(404, "Source not found")
|
||||
result = [scrape_source(conn, dict(s))]
|
||||
else:
|
||||
result = scrape_all_active(conn)
|
||||
return {"results": result, "total_sources": len(result)}
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
@router.post("/bank-forecasts/push-consensus")
|
||||
def push_consensus_endpoint(series_id: str = "", event_date: str = ""):
|
||||
"""Push bank consensus to macro_series_log. Filters optional."""
|
||||
from services.database import get_conn
|
||||
from services.bank_forecast_scraper import push_consensus_to_log, push_all_consensus
|
||||
conn = get_conn()
|
||||
try:
|
||||
if series_id and event_date:
|
||||
val = push_consensus_to_log(conn, series_id, event_date)
|
||||
return {"pushed": [{"series_id": series_id, "event_date": event_date, "consensus": val}]}
|
||||
else:
|
||||
results = push_all_consensus(conn)
|
||||
return {"pushed": results}
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
225
backend/services/bank_forecast_scraper.py
Normal file
225
backend/services/bank_forecast_scraper.py
Normal file
@@ -0,0 +1,225 @@
|
||||
"""
|
||||
Bank forecast scraper — fetches public research pages, extracts numeric forecasts via LLM.
|
||||
Forecasts are stored in bank_forecasts and optionally pushed to macro_series_log as consensus.
|
||||
"""
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from typing import Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Max article text sent to LLM (chars)
|
||||
_MAX_TEXT = 6000
|
||||
|
||||
|
||||
def _fetch_text(url: str, timeout: int = 15) -> str:
|
||||
"""HTTP GET → plain text (strips HTML tags)."""
|
||||
import urllib.request
|
||||
import html
|
||||
req = urllib.request.Request(url, headers={
|
||||
"User-Agent": "Mozilla/5.0 (compatible; GeoOptions-Research/1.0)"
|
||||
})
|
||||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||||
raw = resp.read().decode("utf-8", errors="replace")
|
||||
# Strip HTML tags
|
||||
text = re.sub(r"<[^>]+>", " ", raw)
|
||||
text = html.unescape(text)
|
||||
text = re.sub(r"\s+", " ", text).strip()
|
||||
return text[:_MAX_TEXT]
|
||||
|
||||
|
||||
def _upcoming_events(conn, days_ahead: int = 14) -> list[dict]:
|
||||
"""Return ff_calendar events with series_id coming in the next N days."""
|
||||
today = datetime.now(timezone.utc).date().isoformat()
|
||||
horizon = (datetime.now(timezone.utc).date() + timedelta(days=days_ahead)).isoformat()
|
||||
rows = conn.execute(
|
||||
"""SELECT DISTINCT series_id, event_name, event_date, currency, impact
|
||||
FROM ff_calendar
|
||||
WHERE series_id IS NOT NULL AND series_id != ''
|
||||
AND event_date BETWEEN ? AND ?
|
||||
AND impact IN ('High', 'Medium')
|
||||
ORDER BY event_date""",
|
||||
(today, horizon)
|
||||
).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def _llm_extract(article_text: str, upcoming: list[dict]) -> list[dict]:
|
||||
"""
|
||||
Call Claude to extract numeric forecasts for upcoming events from article text.
|
||||
Returns list of {series_id, event_name, event_date, forecast_value, unit, confidence, snippet}.
|
||||
"""
|
||||
from services.database import get_config as _cfg
|
||||
import anthropic
|
||||
|
||||
api_key = _cfg("anthropic_api_key") or ""
|
||||
if not api_key:
|
||||
logger.warning("[bank_forecast] No anthropic_api_key configured")
|
||||
return []
|
||||
|
||||
if not upcoming:
|
||||
return []
|
||||
|
||||
events_block = "\n".join(
|
||||
f"- {e['event_name']} (series_id={e['series_id']}, release={e['event_date']}, currency={e['currency']})"
|
||||
for e in upcoming
|
||||
)
|
||||
|
||||
prompt = f"""You are extracting economic forecasts from a bank research note.
|
||||
|
||||
Upcoming economic releases:
|
||||
{events_block}
|
||||
|
||||
From the text below, extract any numeric forecasts or expectations for the above indicators.
|
||||
Return a JSON array (no commentary, no markdown). Each item:
|
||||
{{ "series_id": "...", "event_name": "...", "event_date": "YYYY-MM-DD",
|
||||
"forecast_value": <number>, "unit": "%|K|pp|...",
|
||||
"confidence": "high|medium|low",
|
||||
"snippet": "<exact quote from text, max 120 chars>" }}
|
||||
|
||||
Only include entries where a clear numeric forecast is stated. Return [] if nothing found.
|
||||
|
||||
TEXT:
|
||||
{article_text}"""
|
||||
|
||||
client = anthropic.Anthropic(api_key=api_key)
|
||||
msg = client.messages.create(
|
||||
model="claude-haiku-4-5-20251001",
|
||||
max_tokens=1024,
|
||||
messages=[{"role": "user", "content": prompt}]
|
||||
)
|
||||
raw = msg.content[0].text.strip()
|
||||
# Extract JSON array from response
|
||||
m = re.search(r"\[.*\]", raw, re.DOTALL)
|
||||
if not m:
|
||||
return []
|
||||
try:
|
||||
return json.loads(m.group())
|
||||
except Exception as e:
|
||||
logger.warning(f"[bank_forecast] JSON parse error: {e} — raw: {raw[:200]}")
|
||||
return []
|
||||
|
||||
|
||||
def scrape_source(conn, source: dict) -> dict:
|
||||
"""
|
||||
Scrape one bank source: fetch URL, extract forecasts via LLM, save to bank_forecasts.
|
||||
Returns {source_id, name, fetched, extracted, saved, error}.
|
||||
"""
|
||||
sid = source["id"]
|
||||
url = (source.get("url") or "").strip()
|
||||
if not url:
|
||||
return {"source_id": sid, "name": source["name"], "fetched": False,
|
||||
"extracted": 0, "saved": 0, "error": "no URL configured"}
|
||||
|
||||
# Fetch article text
|
||||
try:
|
||||
text = _fetch_text(url)
|
||||
except Exception as e:
|
||||
return {"source_id": sid, "name": source["name"], "fetched": False,
|
||||
"extracted": 0, "saved": 0, "error": str(e)}
|
||||
|
||||
# Get upcoming events to focus extraction
|
||||
upcoming = _upcoming_events(conn)
|
||||
if not upcoming:
|
||||
return {"source_id": sid, "name": source["name"], "fetched": True,
|
||||
"extracted": 0, "saved": 0, "error": "no upcoming events with series_id"}
|
||||
|
||||
# LLM extraction
|
||||
try:
|
||||
forecasts = _llm_extract(text, upcoming)
|
||||
except Exception as e:
|
||||
return {"source_id": sid, "name": source["name"], "fetched": True,
|
||||
"extracted": 0, "saved": 0, "error": f"LLM error: {e}"}
|
||||
|
||||
# Save results
|
||||
saved = 0
|
||||
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
|
||||
for f in forecasts:
|
||||
series_id = f.get("series_id", "").strip()
|
||||
event_date = f.get("event_date", "").strip()
|
||||
forecast_value = f.get("forecast_value")
|
||||
if not series_id or not event_date or forecast_value is None:
|
||||
continue
|
||||
try:
|
||||
conn.execute(
|
||||
"""INSERT INTO bank_forecasts
|
||||
(source_id, series_id, event_name, event_date, forecast_value,
|
||||
unit, confidence, snippet, source_url, extracted_at)
|
||||
VALUES (?,?,?,?,?,?,?,?,?,?)
|
||||
ON CONFLICT(source_id, series_id, event_date) DO UPDATE SET
|
||||
forecast_value=excluded.forecast_value,
|
||||
snippet=excluded.snippet,
|
||||
extracted_at=excluded.extracted_at""",
|
||||
(sid, series_id, f.get("event_name", ""), event_date,
|
||||
float(forecast_value), f.get("unit", ""),
|
||||
f.get("confidence", "medium"), f.get("snippet", "")[:200],
|
||||
url, now)
|
||||
)
|
||||
saved += 1
|
||||
except Exception as e:
|
||||
logger.warning(f"[bank_forecast] save error {sid}/{series_id}: {e}")
|
||||
|
||||
conn.execute("UPDATE bank_forecast_sources SET last_scraped=? WHERE id=?", (now, sid))
|
||||
conn.commit()
|
||||
|
||||
return {"source_id": sid, "name": source["name"], "fetched": True,
|
||||
"extracted": len(forecasts), "saved": saved, "error": None}
|
||||
|
||||
|
||||
def scrape_all_active(conn) -> list[dict]:
|
||||
"""Scrape all active bank sources. Returns list of per-source results."""
|
||||
sources = conn.execute(
|
||||
"SELECT id, name, url, last_scraped FROM bank_forecast_sources WHERE active=1"
|
||||
).fetchall()
|
||||
results = []
|
||||
for s in sources:
|
||||
res = scrape_source(conn, dict(s))
|
||||
results.append(res)
|
||||
logger.info(f"[bank_forecast] {res['name']}: fetched={res['fetched']} saved={res['saved']} err={res.get('error')}")
|
||||
return results
|
||||
|
||||
|
||||
def push_consensus_to_log(conn, series_id: str, event_date: str) -> Optional[float]:
|
||||
"""
|
||||
Compute average of all bank forecasts for (series_id, event_date),
|
||||
push to macro_series_log as source='bank_consensus'.
|
||||
Returns consensus value or None.
|
||||
"""
|
||||
rows = conn.execute(
|
||||
"SELECT forecast_value FROM bank_forecasts WHERE series_id=? AND event_date=? AND forecast_value IS NOT NULL",
|
||||
(series_id, event_date)
|
||||
).fetchall()
|
||||
if not rows:
|
||||
return None
|
||||
|
||||
values = [r[0] for r in rows]
|
||||
consensus = round(sum(values) / len(values), 4)
|
||||
|
||||
# Get event_name from any row
|
||||
meta = conn.execute(
|
||||
"SELECT event_name FROM bank_forecasts WHERE series_id=? AND event_date=? LIMIT 1",
|
||||
(series_id, event_date)
|
||||
).fetchone()
|
||||
event_name = meta[0] if meta else series_id
|
||||
|
||||
from services.macro_series_log import log_if_changed
|
||||
log_if_changed(
|
||||
conn, series_id=series_id, event_name=event_name, event_date=event_date,
|
||||
actual_value=None, forecast_value=consensus, previous_value=None,
|
||||
source="bank_consensus"
|
||||
)
|
||||
return consensus
|
||||
|
||||
|
||||
def push_all_consensus(conn) -> list[dict]:
|
||||
"""Push consensus for every (series_id, event_date) that has bank forecasts."""
|
||||
pairs = conn.execute(
|
||||
"SELECT DISTINCT series_id, event_date FROM bank_forecasts"
|
||||
).fetchall()
|
||||
results = []
|
||||
for series_id, event_date in pairs:
|
||||
val = push_consensus_to_log(conn, series_id, event_date)
|
||||
results.append({"series_id": series_id, "event_date": event_date, "consensus": val})
|
||||
return results
|
||||
@@ -738,6 +738,54 @@ def init_db():
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# ── Bank Forecast Sources ──────────────────────────────────────────────────
|
||||
c.execute("""CREATE TABLE IF NOT EXISTS bank_forecast_sources (
|
||||
id TEXT PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
url TEXT NOT NULL DEFAULT '',
|
||||
active INTEGER NOT NULL DEFAULT 1,
|
||||
last_scraped TEXT,
|
||||
notes TEXT DEFAULT ''
|
||||
)""")
|
||||
# Pre-seed well-known public research sites (URL left blank — user fills in)
|
||||
_default_banks = [
|
||||
('BFS_ING', 'ING Think', ''),
|
||||
('BFS_RABO', 'Rabobank Economics',''),
|
||||
('BFS_COMM', 'Commerzbank Research',''),
|
||||
('BFS_DANSK', 'Danske Research', ''),
|
||||
('BFS_MUFG', 'MUFG Research', ''),
|
||||
('BFS_WELLS', 'Wells Fargo Economics',''),
|
||||
('BFS_BBVA', 'BBVA Research', ''),
|
||||
('BFS_NATIX', 'Natixis Research', ''),
|
||||
]
|
||||
for _bid, _bname, _burl in _default_banks:
|
||||
try:
|
||||
c.execute("INSERT OR IGNORE INTO bank_forecast_sources (id,name,url) VALUES (?,?,?)",
|
||||
(_bid, _bname, _burl))
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
c.execute("""CREATE TABLE IF NOT EXISTS bank_forecasts (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
source_id TEXT NOT NULL REFERENCES bank_forecast_sources(id),
|
||||
series_id TEXT NOT NULL,
|
||||
event_name TEXT NOT NULL DEFAULT '',
|
||||
event_date TEXT NOT NULL,
|
||||
forecast_value REAL,
|
||||
unit TEXT DEFAULT '',
|
||||
confidence TEXT DEFAULT 'medium',
|
||||
snippet TEXT DEFAULT '',
|
||||
source_url TEXT DEFAULT '',
|
||||
extracted_at TEXT NOT NULL DEFAULT (datetime('now')),
|
||||
UNIQUE(source_id, series_id, event_date)
|
||||
)""")
|
||||
try:
|
||||
c.execute("CREATE INDEX IF NOT EXISTS idx_bf_series ON bank_forecasts(series_id, event_date)")
|
||||
c.execute("CREATE INDEX IF NOT EXISTS idx_bf_date ON bank_forecasts(event_date)")
|
||||
c.execute("CREATE INDEX IF NOT EXISTS idx_bf_source ON bank_forecasts(source_id)")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# ── Specialist Desks ───────────────────────────────────────────────────────
|
||||
c.execute("""CREATE TABLE IF NOT EXISTS specialist_reports (
|
||||
id TEXT PRIMARY KEY,
|
||||
|
||||
Reference in New Issue
Block a user