diff --git a/agents/graph.py b/agents/graph.py index 51d902a..3c3aefd 100644 --- a/agents/graph.py +++ b/agents/graph.py @@ -237,6 +237,7 @@ def run_agent( step_budget: int, min_steps: int = 0, min_seconds: float = 0.0, + context_blurb: str = "", checkpointer=None, model_name: str = "", ) -> AgentOutput: @@ -252,9 +253,12 @@ def run_agent( compact_budget = get_settings().compaction_token_budget date_str = datetime.now().astimezone().strftime("%A %Y-%m-%d") + kickoff = RESEARCH_KICKOFF.format(date=date_str, n=persona.target_pick_count) + if context_blurb: # market regime + the agent's own track record (self-learning) + kickoff = f"{context_blurb}\n\n{kickoff}" messages: list = [ SystemMessage(content=persona.system_prompt), - HumanMessage(content=RESEARCH_KICKOFF.format(date=date_str, n=persona.target_pick_count)), + HumanMessage(content=kickoff), ] running_summary = "" reasoning_trace: list[str] = [] diff --git a/execution/run_market_open.py b/execution/run_market_open.py index 9d84b5c..7943830 100644 --- a/execution/run_market_open.py +++ b/execution/run_market_open.py @@ -24,6 +24,7 @@ from execution import broker as broker_mod # noqa: E402 from execution import reconcile as reconcile_mod # noqa: E402 from execution.guardrails import plan_orders # noqa: E402 from orchestration.budget import next_market_trade_date # noqa: E402 +from orchestration.committee import load_verdicts # noqa: E402 from reporting import email_report, report_writer # noqa: E402 from state import db # noqa: E402 from tools.market_data import latest_price # noqa: E402 @@ -69,6 +70,22 @@ def main(force: bool = False, date: str | None = None) -> int: if not outputs: logger.warning("no agent outputs to act on.") return 0 + + # enforce Investment Committee verdicts: veto = drop, reduce = halve size + verdicts = load_verdicts(trade_date) + if verdicts: + for o in outputs: + kept = [] + for p in o.picks: + v = verdicts.get((o.persona_id, p.ticker)) + if v and v["verdict"] == "veto": + logger.info("committee VETO {} {} — {}", o.persona_id, p.ticker, v["reason"]) + continue + if v and v["verdict"] == "reduce": + p.requested_weight_pct = round(p.requested_weight_pct / 2, 4) + logger.info("committee REDUCE {} {} — {}", o.persona_id, p.ticker, v["reason"]) + kept.append(p) + o.picks = kept by_persona = {o.persona_id: o for o in outputs} # 3. fresh prices — equity by ticker, options by contract premium (per share) diff --git a/orchestration/committee.py b/orchestration/committee.py new file mode 100644 index 0000000..526f0fd --- /dev/null +++ b/orchestration/committee.py @@ -0,0 +1,151 @@ +"""The Investment Committee — a Chief Risk Officer pass over the whole night's book. + +After all three agents finalize, a clean-context LLM call reviews every pick against the +market regime, cross-book concentration, and each agent's track record, then issues a +verdict per pick: APPROVE, REDUCE (halve size), or VETO (drop, with reason). The morning +executor enforces the verdicts. A committee memo (markdown) lands in the reports dir so the +dashboard renders it like the other notes. + +This is the layer no retail bot has: the agents don't trade unsupervised — their book has +to clear a risk officer first. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +from langchain_core.messages import HumanMessage, SystemMessage +from loguru import logger + +from agents.schemas import AgentOutput +from config.settings import get_settings +from llm.client import build_llm + +CRO_SYSTEM = ( + "You are the Chief Risk Officer of a small fund. Three analysts submitted tonight's picks. " + "Your job is capital preservation and portfolio coherence — NOT stock picking. Judge each " + "pick on: thesis quality vs the market regime, concentration/overlap across the three books, " + "crowding into one theme, position sizing sanity, and each analyst's recent track record. " + "Be decisive and specific. Output ONLY the requested JSON object." +) + +CRO_PROMPT = ( + "{regime}\n\n" + "ANALYST TRACK RECORDS:\n{records}\n\n" + "TONIGHT'S SUBMITTED PICKS:\n{picks}\n\n" + 'Return a JSON object: {{"memo": "", ' + '"verdicts": [{{"agent": "", "ticker": "", "verdict": "approve|reduce|veto", ' + '"reason": ""}}]}}\n' + "Every submitted pick MUST appear exactly once in verdicts. reduce = halve the size. " + "veto = do not trade it. Veto sparingly — only for real risk problems, not style disagreements." +) + + +def _fmt_picks(outputs: list[AgentOutput]) -> str: + lines = [] + for o in outputs: + for p in o.picks: + inst = p.instrument.value if hasattr(p.instrument, "value") else p.instrument + lines.append( + f"- agent={o.persona_id} {p.ticker} [{inst}/{p.direction.value}] " + f"conv={p.conviction} weight={p.requested_weight_pct} " + f"stop={p.stop_loss_pct} take={p.take_profit_pct} :: {p.thesis[:160]}" + ) + return "\n".join(lines) or "(none)" + + +def _extract_json(text: str) -> dict | None: + t = text.strip().strip("`") + start, end = t.find("{"), t.rfind("}") + if start == -1 or end <= start: + return None + try: + return json.loads(t[start : end + 1]) + except Exception: + return None + + +def run_committee( + outputs: list[AgentOutput], trade_date: str, regime: str, records: dict[str, str] +) -> dict: + """Run the CRO pass. Returns {(agent,ticker): {"verdict","reason"}}; writes memo + verdicts. + + Fail-open: if the model or parsing fails, everything is approved (the guardrails still + apply downstream) — the committee can only make the book safer, never block it entirely. + """ + s = get_settings() + picks_txt = _fmt_picks(outputs) + if picks_txt == "(none)": + logger.info("committee: no picks to review") + return {} + records_txt = "\n".join(f"[{k}]\n{v or '(no history yet)'}" for k, v in records.items()) + + verdicts: dict = {} + memo = "" + try: + resp = build_llm(max_tokens=4096).invoke( + [ + SystemMessage(content=CRO_SYSTEM), + HumanMessage( + content=CRO_PROMPT.format( + regime=regime or "(regime unavailable)", + records=records_txt, + picks=picks_txt, + ) + ), + ] + ) + content = resp.content if isinstance(resp.content, str) else str(resp.content) + data = _extract_json(content) or {} + memo = str(data.get("memo", "")) + for v in data.get("verdicts", []): + if not isinstance(v, dict): + continue + key = (str(v.get("agent", "")).strip(), str(v.get("ticker", "")).strip().upper()) + verdict = str(v.get("verdict", "approve")).lower() + if verdict not in ("approve", "reduce", "veto"): + verdict = "approve" + verdicts[key] = {"verdict": verdict, "reason": str(v.get("reason", ""))[:300]} + except Exception as e: + logger.warning("committee failed (fail-open, all approved): {}", e) + + # persist: verdicts next to the agent JSON (executor reads it), memo into reports + out_dir = s.output_dir / trade_date + out_dir.mkdir(parents=True, exist_ok=True) + (out_dir / "_committee.json").write_text( + json.dumps( + {f"{a}|{t}": v for (a, t), v in verdicts.items()} | {"_memo": memo}, indent=2 + ) + ) + rep_dir = s.reports_dir / trade_date + rep_dir.mkdir(parents=True, exist_ok=True) + vet_lines = "\n".join( + f"- **{t}** ({a}): **{v['verdict'].upper()}** — {v['reason']}" + for (a, t), v in verdicts.items() + ) + (rep_dir / "committee.md").write_text( + f"# Investment Committee Memo — {trade_date}\n\n{memo or '(memo unavailable)'}\n\n" + f"## Verdicts\n{vet_lines or '(all picks approved by default)'}\n" + ) + n_veto = sum(1 for v in verdicts.values() if v["verdict"] == "veto") + n_red = sum(1 for v in verdicts.values() if v["verdict"] == "reduce") + logger.info("committee: {} verdicts ({} veto, {} reduce)", len(verdicts), n_veto, n_red) + return verdicts + + +def load_verdicts(trade_date: str) -> dict: + """Executor-side: {(agent,ticker): {"verdict","reason"}} or {} if no committee ran.""" + path: Path = get_settings().output_dir / trade_date / "_committee.json" + if not path.exists(): + return {} + try: + raw = json.loads(path.read_text()) + return { + tuple(k.split("|", 1)): v + for k, v in raw.items() + if k != "_memo" and isinstance(v, dict) + } + except Exception: + return {} diff --git a/orchestration/regime.py b/orchestration/regime.py new file mode 100644 index 0000000..31abce4 --- /dev/null +++ b/orchestration/regime.py @@ -0,0 +1,67 @@ +"""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 "" diff --git a/orchestration/run_nightly.py b/orchestration/run_nightly.py index c7d6a90..8d23836 100644 --- a/orchestration/run_nightly.py +++ b/orchestration/run_nightly.py @@ -30,8 +30,11 @@ from orchestration.budget import ( # noqa: E402 next_deadline, next_market_trade_date, ) +from orchestration import committee as committee_mod # noqa: E402 from orchestration.checkpoint import Checkpointer # noqa: E402 +from orchestration.regime import market_regime # noqa: E402 from reporting import report_writer # noqa: E402 +from reporting.journal import lessons_for # noqa: E402 def main(compressed: bool = False) -> int: @@ -73,7 +76,13 @@ def main(compressed: bool = False) -> int: min_steps = 4 if compressed else s.per_agent_min_steps min_seconds = 60 if compressed else s.per_agent_min_minutes * 60 + # shared ground truth for all agents tonight (computed, deterministic) + regime = market_regime() + if regime: + logger.info("regime: {}", regime[:140]) + outputs = [] + records: dict[str, str] = {} for i, persona in enumerate(PERSONAS): now = time.time() if now >= hard_stop: @@ -85,9 +94,12 @@ def main(compressed: bool = False) -> int: if not server.server_up(): # recover if a prior agent left oMLX unresponsive logger.warning("oMLX unreachable before {}; recovering...", persona.id) server.ensure_ready(warm=False) + lessons = lessons_for(persona.id) # self-learning: past picks marked to market + records[persona.id] = lessons + blurb = "\n\n".join(x for x in (regime, lessons) if x) out = run_agent(persona, deadline_ts=deadline, step_budget=step_budget, min_steps=min_steps, min_seconds=min_seconds, - checkpointer=cp, model_name=model_name) + context_blurb=blurb, checkpointer=cp, model_name=model_name) cp.write(out) outputs.append(out) manifest.agents.append(persona.id) @@ -96,6 +108,12 @@ def main(compressed: bool = False) -> int: except Exception as e: logger.warning("agent report failed for {}: {}", persona.id, e) + # Investment Committee: risk-officer pass over the whole book (verdicts enforce at open) + try: + committee_mod.run_committee(outputs, trade_date, regime, records) + except Exception as e: + logger.warning("committee failed (fail-open): {}", e) + # combined daily report (execution results appended in the morning by run_market_open) try: daily = report_writer.write_daily_report(trade_date, outputs) diff --git a/reporting/journal.py b/reporting/journal.py new file mode 100644 index 0000000..52b117e --- /dev/null +++ b/reporting/journal.py @@ -0,0 +1,64 @@ +"""Self-learning trade journal — each agent sees its own past picks AND how they worked out +before researching tonight. Losses and wins become explicit lessons instead of being forgotten. + +Joins past theses with their execution price (orders table) and marks each to market via the +current price. Compact text block injected into the agent's kickoff. +""" + +from __future__ import annotations + +from loguru import logger + +from config.settings import get_settings +from state import db as statedb +from tools.market_data import latest_price + +_MAX_ROWS = 10 + + +def lessons_for(persona_id: str) -> str: + """Past-pick scorecard for one agent. Empty string if no history/DB yet.""" + s = get_settings() + if not s.db_path.exists(): + return "" + try: + con = statedb.connect(s.db_path) + rows = con.execute( + """ + SELECT t.trade_date, t.ticker, t.direction, t.conviction, t.thesis, + o.est_price, o.status + FROM theses t + LEFT JOIN orders o + ON o.trade_date = t.trade_date AND o.agent = t.agent AND o.ticker = t.ticker + WHERE t.agent = ? + ORDER BY t.trade_date DESC, t.conviction DESC + LIMIT ? + """, + (persona_id, _MAX_ROWS), + ).fetchall() + con.close() + except Exception as e: + logger.warning("journal query failed for {}: {}", persona_id, e) + return "" + if not rows: + return "" + + lines = [] + for r in rows: + entry = r["est_price"] + outcome = "not executed" + if entry: + px = latest_price(r["ticker"]) + if px: + chg = (px / entry - 1) * 100 + if (r["direction"] or "long") == "short": + chg = -chg + outcome = f"{chg:+.1f}% since {r['trade_date']} (entry ~{entry:.2f})" + lines.append( + f"- {r['trade_date']} {r['ticker']} {r['direction']} conv={r['conviction']}: {outcome}" + f" | thesis was: {(r['thesis'] or '')[:100]}" + ) + return ( + "YOUR TRACK RECORD (mark-to-market of your past picks — learn from it; double down on " + "what worked, avoid repeating what failed):\n" + "\n".join(lines) + )