Aryan SinghandClaude Opus 4.8
4beee3dc82
- regime.py: computed SPY/VIX/sector-momentum snapshot injected into every kickoff - journal.py: each agent sees its own past picks marked-to-market (learns from wins/losses) - committee.py: Chief-Risk-Officer LLM pass over the whole book — approve/reduce/veto per pick with reasons; memo lands in reports; executor enforces verdicts at open Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016DvdRZkyCy9NJpHYDdPqmS
68 lines
2.4 KiB
Python
68 lines
2.4 KiB
Python
"""Market regime snapshot — computed in code (deterministic, no LLM), injected into every
|
|
agent's kickoff so all three personas share the same ground truth about the tape.
|
|
|
|
Covers: SPY trend (5/20/50d), VIX level, and sector ETF momentum ranking (20d).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from loguru import logger
|
|
|
|
_SECTORS = {
|
|
"XLK": "Tech", "XLF": "Financials", "XLE": "Energy", "XLV": "Health",
|
|
"XLY": "Cons.Disc", "XLP": "Staples", "XLI": "Industrials", "XLU": "Utilities",
|
|
"XLB": "Materials", "XLC": "Comms",
|
|
}
|
|
|
|
|
|
def _pct(series, days: int) -> float | None:
|
|
try:
|
|
if len(series) <= days:
|
|
return None
|
|
return float((series.iloc[-1] / series.iloc[-1 - days] - 1) * 100)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def market_regime() -> str:
|
|
"""One compact paragraph describing the current tape. Empty string on failure."""
|
|
try:
|
|
import yfinance as yf
|
|
|
|
data = yf.download(
|
|
["SPY", "^VIX", *_SECTORS], period="4mo", interval="1d",
|
|
progress=False, group_by="ticker", threads=True,
|
|
)
|
|
spy = data["SPY"]["Close"].dropna()
|
|
r5, r20, r50 = _pct(spy, 5), _pct(spy, 20), _pct(spy, 50)
|
|
trend = "uptrend" if (r20 or 0) > 0 and (r50 or 0) > 0 else (
|
|
"downtrend" if (r20 or 0) < 0 and (r50 or 0) < 0 else "mixed/chop")
|
|
|
|
try:
|
|
vix = float(data["^VIX"]["Close"].dropna().iloc[-1])
|
|
vol = "low" if vix < 15 else ("elevated" if vix < 25 else "HIGH — risk-off")
|
|
vix_s = f"VIX {vix:.1f} ({vol} volatility)"
|
|
except Exception:
|
|
vix_s = "VIX unavailable"
|
|
|
|
sect = []
|
|
for etf, name in _SECTORS.items():
|
|
try:
|
|
r = _pct(data[etf]["Close"].dropna(), 20)
|
|
if r is not None:
|
|
sect.append((name, r))
|
|
except Exception:
|
|
continue
|
|
sect.sort(key=lambda x: x[1], reverse=True)
|
|
leaders = ", ".join(f"{n} {r:+.1f}%" for n, r in sect[:3])
|
|
laggards = ", ".join(f"{n} {r:+.1f}%" for n, r in sect[-3:])
|
|
|
|
return (
|
|
f"MARKET REGIME (computed now): SPY {trend} — 5d {r5:+.1f}%, 20d {r20:+.1f}%, "
|
|
f"50d {r50:+.1f}%. {vix_s}. Sector momentum (20d): leaders {leaders}; "
|
|
f"laggards {laggards}."
|
|
)
|
|
except Exception as e:
|
|
logger.warning("regime snapshot failed: {}", e)
|
|
return ""
|