Files
OpenFin/backend/services/ma_analyzer.py
OpenSquared 16ccc7c2c7 feat: cycle
2026-07-15 14:59:11 +02:00

573 lines
21 KiB
Python

"""
MA Analyzer — Bootstrap historical market periods from price-based MA signals.
Fetches 5y daily prices for 5 key underlyings, detects MA ruptures (golden/death
cross, MA100 slope changes, MA20 direction changes), enriches candidates with
GPT-4o-mini, and persists them via save_market_event().
"""
import json
import logging
import os
import time
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional, Tuple
import numpy as np
import pandas as pd
import yfinance as yf
from services.database import count_market_events, save_market_event
logger = logging.getLogger(__name__)
UNDERLYINGS = [
{"ticker": "EURUSD=X", "name": "EUR/USD", "category": "market", "assets": ["EUR/USD", "USD", "EUR"]},
{"ticker": "BZ=F", "name": "Brent Crude", "category": "market", "assets": ["CL", "BZ", "Oil"]},
{"ticker": "GC=F", "name": "Gold", "category": "market", "assets": ["GC", "Gold", "GLD"]},
{"ticker": "^GSPC", "name": "S&P 500", "category": "market", "assets": ["SPX", "SPY", "QQQ"]},
{"ticker": "^TNX", "name": "US 10Y Yield", "category": "market", "assets": ["TNX", "TLT", "Bonds"]},
]
MA_PERIODS = [20, 50, 100, 200]
CROSS_CONFIRM_DAYS = 5
SLOPE_WINDOW = 10
DEDUP_DAYS = 30
MAX_CANDIDATES = 80
GPT_BATCH_SIZE = 5
def _fetch_prices() -> Dict[str, pd.Series]:
"""Download 5y daily close prices for all underlyings. Returns ticker → pd.Series."""
tickers = [u["ticker"] for u in UNDERLYINGS]
prices: Dict[str, pd.Series] = {}
try:
raw = yf.download(tickers, period="5y", interval="1d", progress=False, auto_adjust=True)
if raw.empty:
raise ValueError("Empty download result")
close = raw["Close"] if isinstance(raw.columns, pd.MultiIndex) else raw
for ticker in tickers:
if ticker in close.columns:
s = close[ticker].dropna()
if not s.empty:
prices[ticker] = s
logger.info(f"[MA] {ticker}: {len(s)} bars")
except Exception as e:
logger.warning(f"[MA] Batch download failed ({e}), falling back to individual")
for ticker in tickers:
try:
df = yf.download(ticker, period="5y", interval="1d", progress=False, auto_adjust=True)
if df.empty:
continue
if isinstance(df.columns, pd.MultiIndex):
df.columns = df.columns.get_level_values(0)
s = df["Close"].dropna()
if not s.empty:
prices[ticker] = s
logger.info(f"[MA] {ticker}: {len(s)} bars")
except Exception as e2:
logger.error(f"[MA] Failed to fetch {ticker}: {e2}")
return prices
def _compute_mas(prices: pd.Series) -> pd.DataFrame:
"""Return DataFrame with Close + MA20/50/100/200 columns."""
df = prices.to_frame(name="Close")
for p in MA_PERIODS:
df[f"MA{p}"] = df["Close"].rolling(p).mean()
return df
def _date_str(ts) -> str:
if hasattr(ts, "strftime"):
return ts.strftime("%Y-%m-%d")
return str(ts)[:10]
def _price_move_pct(df: pd.DataFrame, start_idx: int, end_idx: int) -> float:
"""Absolute % move of Close between two integer iloc positions."""
try:
p0 = float(df["Close"].iloc[start_idx])
p1 = float(df["Close"].iloc[end_idx])
if p0 == 0:
return 0.0
return abs((p1 - p0) / p0 * 100)
except Exception:
return 0.0
def _detect_crossovers(df: pd.DataFrame, underlying: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Detect MA50/MA200 golden cross and death cross with 5-day confirmation."""
candidates = []
ma50 = df["MA50"]
ma200 = df["MA200"]
valid = df.dropna(subset=["MA50", "MA200"])
if len(valid) < CROSS_CONFIRM_DAYS + 2:
return candidates
idx_list = valid.index.tolist()
i = 1
while i < len(idx_list) - CROSS_CONFIRM_DAYS:
prev_i = idx_list[i - 1]
curr_i = idx_list[i]
prev_above = ma50[prev_i] > ma200[prev_i]
curr_above = ma50[curr_i] > ma200[curr_i]
if prev_above == curr_above:
i += 1
continue
is_golden = curr_above # MA50 just crossed above MA200
# Confirm: next CROSS_CONFIRM_DAYS all maintain the new relationship
confirm_ok = True
for k in range(1, CROSS_CONFIRM_DAYS + 1):
if i + k >= len(idx_list):
confirm_ok = False
break
future_i = idx_list[i + k]
if is_golden:
if not (ma50[future_i] > ma200[future_i]):
confirm_ok = False
break
else:
if not (ma50[future_i] < ma200[future_i]):
confirm_ok = False
break
if confirm_ok:
# start_date = date of the confirming bar (i + CROSS_CONFIRM_DAYS)
confirmed_i = idx_list[i + CROSS_CONFIRM_DAYS]
start_iloc = valid.index.get_loc(confirmed_i)
event_type = "golden_cross" if is_golden else "death_cross"
name_prefix = "Golden Cross" if is_golden else "Death Cross"
candidates.append({
"_type": event_type,
"_underlying": underlying,
"_start_iloc": start_iloc,
"start_date": _date_str(confirmed_i),
"end_date": None,
"level": "medium",
"category": "market",
"_move_magnitude": _price_move_pct(df, max(0, start_iloc - 20), start_iloc),
"_name_hint": f"{underlying['name']} {name_prefix} MA50/MA200",
})
i += CROSS_CONFIRM_DAYS + 1
else:
i += 1
return candidates
def _slope(series: pd.Series, window: int = SLOPE_WINDOW) -> pd.Series:
"""Percentage slope over window: (val[t] - val[t-window]) / val[t-window] * 100."""
return (series - series.shift(window)) / series.shift(window) * 100
def _detect_ma100_slope_changes(df: pd.DataFrame, underlying: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Detect MA100 direction reversals (slope sign flips sustained over 5+ bars)."""
candidates = []
valid = df.dropna(subset=["MA100"])
if len(valid) < SLOPE_WINDOW + 10:
return candidates
slope = _slope(valid["MA100"], SLOPE_WINDOW)
sign = np.sign(slope)
idx_list = valid.index.tolist()
i = 1
while i < len(idx_list) - 5:
prev_sign = sign[idx_list[i - 1]]
curr_sign = sign[idx_list[i]]
if prev_sign == 0 or curr_sign == 0 or prev_sign == curr_sign:
i += 1
continue
# Check 5-bar confirmation
confirm_ok = all(
sign[idx_list[i + k]] == curr_sign
for k in range(1, 6)
if i + k < len(idx_list)
)
if not confirm_ok:
i += 1
continue
start_idx = valid.index.get_loc(idx_list[i])
# Find end: next sign flip
end_date = None
duration_days = 0
for j in range(i + 1, len(idx_list)):
if sign[idx_list[j]] == -curr_sign:
end_date = _date_str(idx_list[j])
duration_days = (
datetime.strptime(end_date, "%Y-%m-%d") -
datetime.strptime(_date_str(idx_list[i]), "%Y-%m-%d")
).days
break
direction = "Bullish" if curr_sign > 0 else "Bearish"
level = "long" if duration_days > 90 or end_date is None else "medium"
candidates.append({
"_type": "ma100_slope_change",
"_underlying": underlying,
"_start_iloc": start_idx,
"start_date": _date_str(idx_list[i]),
"end_date": end_date,
"level": level,
"category": "market",
"_move_magnitude": abs(float(slope[idx_list[i]])),
"_name_hint": f"{underlying['name']} MA100 {direction} Trend",
})
# Skip to next potential flip zone
if end_date:
while i < len(idx_list) and _date_str(idx_list[i]) < end_date:
i += 1
else:
break
return candidates
def _detect_ma20_direction_changes(df: pd.DataFrame, underlying: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Detect strong MA20 direction changes (> 2% in 10 days)."""
candidates = []
valid = df.dropna(subset=["MA20"])
if len(valid) < SLOPE_WINDOW + 10:
return candidates
slope = _slope(valid["MA20"], SLOPE_WINDOW)
THRESHOLD = 2.0
idx_list = valid.index.tolist()
last_added_date = None
for i in range(1, len(idx_list) - 5):
prev_slope = float(slope[idx_list[i - 1]])
curr_slope = float(slope[idx_list[i]])
if abs(curr_slope) < THRESHOLD:
continue
if np.sign(prev_slope) == np.sign(curr_slope):
continue
start_date = _date_str(idx_list[i])
# Dedup: skip if too close to last added
if last_added_date:
gap = (
datetime.strptime(start_date, "%Y-%m-%d") -
datetime.strptime(last_added_date, "%Y-%m-%d")
).days
if gap < DEDUP_DAYS:
continue
# Find end: slope returns to < 0.5 or reverses
end_date = None
duration_days = 0
for j in range(i + 1, len(idx_list)):
future_slope = float(slope[idx_list[j]])
if abs(future_slope) < 0.5 or np.sign(future_slope) != np.sign(curr_slope):
end_date = _date_str(idx_list[j])
duration_days = (
datetime.strptime(end_date, "%Y-%m-%d") -
datetime.strptime(start_date, "%Y-%m-%d")
).days
break
direction = "Bullish" if curr_slope > 0 else "Bearish"
level = "short" if (end_date and duration_days < 30) else "medium"
candidates.append({
"_type": "ma20_direction_change",
"_underlying": underlying,
"_start_iloc": valid.index.get_loc(idx_list[i]),
"start_date": start_date,
"end_date": end_date,
"level": level,
"category": "market",
"_move_magnitude": abs(curr_slope),
"_name_hint": f"{underlying['name']} MA20 {direction} Swing",
})
last_added_date = start_date
return candidates
def _dedup_candidates(candidates: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Remove near-duplicate events (same underlying + type, dates < 30 days apart).
Keep the one with the largest _move_magnitude."""
kept: List[Dict[str, Any]] = []
for ev in candidates:
ticker = ev["_underlying"]["ticker"]
ev_type = ev["_type"]
ev_start = ev["start_date"]
duplicate_of = None
for k in kept:
if k["_underlying"]["ticker"] != ticker:
continue
if k["_type"] != ev_type:
continue
gap = abs(
(datetime.strptime(ev_start, "%Y-%m-%d") -
datetime.strptime(k["start_date"], "%Y-%m-%d")).days
)
if gap < DEDUP_DAYS:
duplicate_of = k
break
if duplicate_of is None:
kept.append(ev)
elif ev.get("_move_magnitude", 0) > duplicate_of.get("_move_magnitude", 0):
kept.remove(duplicate_of)
kept.append(ev)
return kept
def _assign_end_dates(candidates: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""For events without an end_date, set it to the next same-type event on same underlying."""
for i, ev in enumerate(candidates):
if ev.get("end_date"):
continue
ticker = ev["_underlying"]["ticker"]
ev_type = ev["_type"]
ev_start = ev["start_date"]
# Find the next event of same type/underlying that comes after
next_start = None
for j, other in enumerate(candidates):
if i == j:
continue
if other["_underlying"]["ticker"] != ticker:
continue
if other["_type"] != ev_type:
continue
if other["start_date"] > ev_start:
if next_start is None or other["start_date"] < next_start:
next_start = other["start_date"]
ev["end_date"] = next_start # None if it's the last (ongoing)
return candidates
def _build_gpt_prompt(batch: List[Dict[str, Any]], df_map: Dict[str, pd.DataFrame]) -> str:
items = []
for i, ev in enumerate(batch):
underlying = ev["_underlying"]
ticker = underlying["ticker"]
df = df_map.get(ticker)
start_price = end_price = None
if df is not None:
try:
start_idx = df.index.searchsorted(pd.Timestamp(ev["start_date"]))
start_price = float(df["Close"].iloc[min(start_idx, len(df) - 1)])
except Exception:
pass
if ev.get("end_date"):
try:
end_idx = df.index.searchsorted(pd.Timestamp(ev["end_date"]))
end_price = float(df["Close"].iloc[min(end_idx, len(df) - 1)])
except Exception:
pass
items.append({
"index": i,
"underlying": underlying["name"],
"type": ev["_type"],
"start_date": ev["start_date"],
"end_date": ev.get("end_date"),
"level": ev["level"],
"name_hint": ev["_name_hint"],
"price_start": round(start_price, 4) if start_price else None,
"price_end": round(end_price, 4) if end_price else None,
})
return f"""You are a senior macro analyst specializing in technical market regimes.
Analyze these {len(batch)} market events detected via Moving Average signals.
For each, return a JSON object with:
- index: (same as input)
- name: concise English event name (ex: "S&P 500 Golden Cross Bull Trend")
- description: 2-3 sentences explaining market context and significance
- market_impact: 1 sentence on trading/volatility impact
- affected_assets: list of 3-6 ticker symbols most affected
- impact_score: float 0.0-1.0 (0=minor technical, 1=major structural)
Events to analyze:
{json.dumps(items, indent=2)}
Return a JSON array of {len(batch)} objects, one per input event, in the same order.
Array only, no wrapping key."""
def _enrich_with_gpt(candidates: List[Dict[str, Any]], df_map: Dict[str, pd.DataFrame]) -> List[Dict[str, Any]]:
"""Call GPT-4o-mini in batches of GPT_BATCH_SIZE to enrich candidates."""
api_key = os.environ.get("OPENAI_API_KEY", "")
if not api_key:
logger.warning("[MA] No OPENAI_API_KEY — skipping GPT enrichment")
for ev in candidates:
ev["name"] = ev["_name_hint"]
ev["description"] = f"MA signal detected on {ev['_underlying']['name']}."
ev["market_impact"] = "Technical price action signal."
ev["affected_assets"] = ev["_underlying"]["assets"]
ev["impact_score"] = 0.4
return candidates
from openai import OpenAI
client = OpenAI(api_key=api_key)
for batch_start in range(0, len(candidates), GPT_BATCH_SIZE):
batch = candidates[batch_start: batch_start + GPT_BATCH_SIZE]
try:
prompt = _build_gpt_prompt(batch, df_map)
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0.3,
max_tokens=1200,
)
raw = response.choices[0].message.content.strip()
# Strip markdown code block if present
if raw.startswith("```"):
raw = raw.split("```")[1]
if raw.startswith("json"):
raw = raw[4:]
results = json.loads(raw)
for item in results:
idx = item.get("index")
if idx is not None and 0 <= idx < len(batch):
ev = batch[idx]
ev["name"] = item.get("name", ev["_name_hint"])
ev["description"] = item.get("description", "")
ev["market_impact"] = item.get("market_impact", "")
ev["affected_assets"] = item.get("affected_assets", ev["_underlying"]["assets"])
ev["impact_score"] = float(item.get("impact_score", 0.4))
except Exception as e:
logger.error(f"[MA] GPT batch {batch_start // GPT_BATCH_SIZE + 1} failed: {e}")
for ev in batch:
if "name" not in ev:
ev["name"] = ev["_name_hint"]
ev["description"] = f"MA signal on {ev['_underlying']['name']}."
ev["market_impact"] = "Technical signal."
ev["affected_assets"] = ev["_underlying"]["assets"]
ev["impact_score"] = 0.4
time.sleep(1)
# Ensure all events have required fields
for ev in candidates:
if "name" not in ev:
ev["name"] = ev["_name_hint"]
ev["description"] = f"MA signal on {ev['_underlying']['name']}."
ev["market_impact"] = "Technical signal."
ev["affected_assets"] = ev["_underlying"]["assets"]
ev["impact_score"] = 0.4
return candidates
def _to_event_dict(ev: Dict[str, Any]) -> Dict[str, Any]:
"""Convert internal candidate to the format expected by save_market_event."""
return {
"name": ev["name"],
"start_date": ev["start_date"],
"end_date": ev.get("end_date"),
"level": ev["level"],
"category": ev.get("category", "market"),
"description": ev.get("description", ""),
"market_impact": ev.get("market_impact", ""),
"affected_assets": ev.get("affected_assets", []),
"impact_score": ev.get("impact_score", 0.4),
"absorption_pct": None,
"relevant_indicators": [],
"origin": "bootstrap_ma",
}
async def bootstrap_ma_events(force: bool = False) -> Dict[str, Any]:
"""
Main entrypoint. Detects MA ruptures, enriches with GPT, saves to DB.
Idempotent by default: skips if DB already has > 100 events. Pass force=True
to bypass this guard (same convention as bootstrap_macro_events/bootstrap_eco_events).
Returns {"detected": N, "saved": N, "skipped": N}.
"""
existing = count_market_events()
if existing > 100 and not force:
logger.info(f"[MA] DB already has {existing} events — skipping bootstrap")
return {"detected": 0, "saved": 0, "skipped": existing}
logger.info("[MA] Fetching price data...")
prices = _fetch_prices()
if not prices:
logger.error("[MA] No price data fetched — aborting")
return {"detected": 0, "saved": 0, "skipped": 0}
df_map: Dict[str, pd.DataFrame] = {}
all_candidates: List[Dict[str, Any]] = []
for underlying in UNDERLYINGS:
ticker = underlying["ticker"]
if ticker not in prices:
logger.warning(f"[MA] No data for {ticker}, skipping")
continue
try:
df = _compute_mas(prices[ticker])
df_map[ticker] = df
cross_evs = _detect_crossovers(df, underlying)
slope_evs = _detect_ma100_slope_changes(df, underlying)
swing_evs = _detect_ma20_direction_changes(df, underlying)
logger.info(
f"[MA] {underlying['name']}: crossovers={len(cross_evs)}, "
f"slope={len(slope_evs)}, swing={len(swing_evs)}"
)
all_candidates.extend(cross_evs + slope_evs + swing_evs)
except Exception as e:
logger.error(f"[MA] Detection failed for {underlying['name']}: {e}")
logger.info(f"[MA] Raw candidates: {len(all_candidates)}")
# Dedup + assign end dates
candidates = _dedup_candidates(all_candidates)
candidates = _assign_end_dates(candidates)
# Sort by start_date ascending for GPT context coherence
candidates.sort(key=lambda x: x["start_date"])
# Cap at MAX_CANDIDATES (keep most significant by _move_magnitude)
if len(candidates) > MAX_CANDIDATES:
candidates.sort(key=lambda x: x.get("_move_magnitude", 0), reverse=True)
candidates = candidates[:MAX_CANDIDATES]
candidates.sort(key=lambda x: x["start_date"])
detected = len(candidates)
logger.info(f"[MA] After dedup/cap: {detected} candidates — enriching with GPT...")
candidates = _enrich_with_gpt(candidates, df_map)
saved = 0
skipped_save = 0
for ev in candidates:
try:
save_market_event(_to_event_dict(ev))
saved += 1
except Exception as e:
logger.error(f"[MA] Failed to save '{ev.get('name', '?')}': {e}")
skipped_save += 1
logger.info(f"[MA] Bootstrap complete: detected={detected} saved={saved} skipped={skipped_save}")
return {"detected": detected, "saved": saved, "skipped": skipped_save}