"""Persona brief, PM anchors and the summary prompt — built from AnalysisState only. Nothing here reads a legacy dict. Every figure appears as a claim token with its value and label, so a model that wants to cite a number cites the claim, and the renderer later substitutes the frozen value. Qualitative context (headlines, the call transcript, filings) is passed as text evidence and is never a number of record. """ from __future__ import annotations from collections.abc import Mapping import json import os import sys import catalog as C from catalog import persona_catalog_text def _t(cid, cat): return f"[[{cid}]]" if cid in cat else "unavailable" def token_header(cat, own_return=True, state=None): from catalog import persona_comparison_text cmp_txt = persona_comparison_text(state, cat) if state is not None else "" return ("RULES FOR NUMBERS (these override any instruction below that asks you to cite a number):\n" + persona_catalog_text(cat, state) + ("\n" + cmp_txt if cmp_txt else "") + ("\nYour own scenario return, once you state it in the JSON field, may be cited as [[OWN_RETURN]]." if own_return else "")) def _engine_block(s): e = s.valuation.get("engine") or {} t = e.get("terms") or {} forb = e.get("forbidden_renderer_terms") or [] return (f"VALUATION METHOD — {e.get('display_name')}. Call the intrinsic estimate \"{s.valuation.get('intrinsic_label') or t.get('intrinsic', 'intrinsic value')}\"" + (f" and its cross-check \"{t.get('cross_check')}\"" if t.get("cross_check") and not s.valuation.get("same_model") else "") + f"; the discount rate is \"{t.get('discount_rate', 'discount rate')}\"." + (f" These terms are wrong for this company and must not appear: {', '.join(forb)}." if forb else "")) def _quant_block(s): pm = ((s.quant.get("price_model") or {}).get("reliability") or {}) pr = ((s.quant.get("panel_rank") or {}).get("reliability") or {}) # a hidden model is named with its state only: none of its statistics reach the author (evidence.py) lines = [f"QUANTITATIVE MODELS — single-stock price model: {pm.get('state')}" + (f" ({pm.get('basis')})" if pm.get("persona_visibility") else "") + ". " + ("Not usable as evidence: do not cite it, and do not let it move your view." if not pm.get("persona_visibility") else f"You may describe it only as '{pm.get('allowed_claim_strength')}'.")] lines.append(f"Cross-sectional panel rank: {pr.get('state')}" + (f" ({pr.get('basis')})" if pr.get("persona_visibility") else "") + ". " + ("Not usable as evidence." if not pr.get("persona_visibility") else f"Usable only as {pr.get('allowed_claim_strength')}: a machine-learning model ranks the stock above " "[[ML_RANK_SHARE]] of the [[ML_RANK_UNIVERSE]] largest US companies (decile [[ML_RANK_DECILE]]) for " "12-month return relative to them. It is a ranking, not a price target.")) return "\n".join(lines) def _peer_words(s): """Peer bands only from relative-valuation claims whose comparison is allowed (same multiple, period basis, accounting basis and perimeter); a withheld subject yields no band and no rich/cheap language.""" pc = s.pricing.get("peers") or {} rel = s.pricing.get("relative") or {} pct = {k: v.get("percentile") for k, v in rel.items() if v.get("comparison_allowed")} if rel else (pc.get("percentiles") or {}) if not pc.get("peers"): return "PEERS — no peer table available." if not pct: return (f"PEERS ({pc.get('sector') or 'sector'}) — the company's own trailing multiples are withheld " f"({pc.get('subject_withheld') or 'not comparable'}), so no comparison with peers is available: do not " "describe the stock as rich, cheap, at a premium or at a discount to peers.") def band(p): if p is None: return None return ("near the bottom" if p < 20 else "below the median" if p < 45 else "around the median" if p <= 55 else "above the median" if p <= 80 else "near the top") parts = [f"{k.replace('_', ' ')} {band(v)}" for k, v in pct.items() if band(v)] names = ", ".join(p.get("ticker") for p in pc["peers"][:8] if p.get("ticker")) return f"PEERS ({pc.get('sector') or 'sector'}; {names}) — relative to peers the company's " + "; ".join(parts) + "." def _author_basis(rec): """Where the trailing figures come from, in reader terms (never the pipeline's reconciliation vocabulary), from the trailing figures' own quality state (HistoricalMetricQuality): an unchecked quarter is said to be unchecked.""" q = (rec or {}).get("ttm_quality") or {} st, unv = q.get("status"), list(q.get("unverified_periods") or []) if st == "SEC_RECONCILED": return "the company's SEC filings" if st == "PARTIALLY_RECONCILED": return (("the company's SEC filings through " + q["verified_through"] if q.get("verified_through") else "the company's SEC filings for some quarters") + (f"; the quarters ended {' and '.join(unv)} come from the data provider and were not checked against a filing" if len(unv) > 1 else f"; the quarter ended {' and '.join(unv)} comes from the data provider and was not checked against a filing")) if st in ("PROVIDER_ONLY", "UNVERIFIED"): return "the data provider's quarterly statements, not checked against SEC filings" return {"RECONCILED": "the company's SEC filings", "SEC_ONLY": "the company's SEC filings"}.get( (rec or {}).get("status"), "the company's reported quarterly statements") def _forecast_lineage(s): """What the engine forecast is, from its typed consensus anchor: a figure SET from consensus is not called a model estimate, and it is not compared with consensus (relations.py publishes no such relation).""" anc = s.forecast.get("consensus_anchor") or {} full = [n for k, n in (("revenue", "revenue"), ("eps", "EPS")) if isinstance(anc.get(k), Mapping) and float(anc[k].get("model_weight", 1.0)) <= 0.0] part = [n for k, n in (("revenue", "revenue"), ("eps", "EPS")) if isinstance(anc.get(k), Mapping) and 0.0 < float(anc[k].get("model_weight", 1.0)) < 1.0] out = "next 12 months; " if full: out += (" and ".join(full) + " are set from sell-side consensus, not an independent model estimate, so they are " "never compared with consensus") else: out += "a model estimate, not guidance or consensus" if part: out += "; " + " and ".join(part) + " shrunk toward sell-side consensus" return out def persona_brief(s, cat): """The shared brief for the neutral, bear and bull seats.""" rec = s.historical.get("reconciliation") or {} fc = s.forecast.get("coherence") or {} stry = s.valuation.get("story") or {} lines = [ f"COMPANY — {s.identity.get('company_name')} ({s.identity.ticker}); {s.issuer.get('issuer_class')}; " f"{'foreign private issuer' if s.issuer.get('is_fpi') else 'US domestic filer'}; reports under {s.issuer.get('accounting_standard')}" + ("; quoted as ADS" if s.issuer.get("is_adr") else "") + ".", _engine_block(s), # the reconciliation's own basis text and notes (provider differences, unverified quarters) describe THIS REPORT'S # data pipeline, not the company: authors get only the reader-facing source, derived from the typed status f"SOURCE — trailing figures come from {_author_basis(rec)}.", f"MARKET — price {_t('PRICE', cat)}; market cap {_t('MARKET_CAP', cat)}; momentum {s.value('market.momentum.state') or 'unavailable'} " f"(shares {_t('RETURN_3M_PHRASE', cat)} and {_t('RETURN_12M_PHRASE', cat)}; 200-day average {_t('SMA200', cat)}); " f"dividend yield {_t('DIVIDEND_YIELD', cat)}.", f"TRAILING FUNDAMENTALS — revenue {_t('REVENUE_TTM', cat)}; operating margin {_t('OPERATING_MARGIN_TTM', cat)}; " f"net margin {_t('NET_MARGIN_TTM', cat)}; adjusted EPS {_t('EPS_ADJUSTED_TTM', cat)}; reported EPS {_t('EPS_GAAP_TTM', cat)}" + (f" (adjusted ÷ reported, like for like: {_t('ADJ_GAAP_BRIDGE', cat)})" if "ADJ_GAAP_BRIDGE" in cat else "") + "; " f"earnings beats {_t('BEAT_RATE', cat)} of the last {s.historical.get('beat_window') or 'few'} quarters.", f"SELL-SIDE — consensus EPS {_t('CONSENSUS_EPS_FY1', cat)} this fiscal year and {_t('CONSENSUS_EPS_FY2', cat)} next " f"({_t('CONSENSUS_ANALYSTS', cat)} contributors); consensus revenue {_t('CONSENSUS_REVENUE_FY1', cat)}; " f"mean price target {_t('ANALYST_TARGET', cat)}; {_t('REL_TARGET_PRICE', cat)}.", f"ENGINE FORECAST ({_forecast_lineage(s)}) — revenue {_t('FORECAST_REVENUE', cat)} " f"({_t('FORECAST_REVENUE_GROWTH', cat)}); operating margin {_t('FORECAST_OPERATING_MARGIN', cat)}; net margin " f"{_t('FORECAST_NET_MARGIN', cat)}; EPS {_t('FORECAST_EPS', cat)} ({_t('FORECAST_EPS_GROWTH', cat)} on the same adjusted basis)" + "".join(f"; {_t(r, cat)}" for r in ("REL_FORECAST_EPS_CONSENSUS", "REL_ENGINE_REVENUE_CONSENSUS") if r in cat) + ".", f"VALUATION — {s.valuation.get('intrinsic_label') or 'intrinsic value'} {_t('INTRINSIC_VALUE', cat)} " f"({_t('REL_PRICE_INTRINSIC', cat)})" + (f"; cross-check {_t('MECHANICAL_VALUE', cat)} ({_t('REL_PRICE_MECHANICAL', cat)})" if "MECHANICAL_VALUE" in cat else "") + ".", ] import guidance as _gd lines.append("COMPANY GUIDANCE — " + _gd.persona_block(s.guidance.get("company"))) seg = s.segments.get("canonical") or {} if seg.get("status") == "RECONCILED" and "SEGMENTS_COVERAGE" in cat: per = s.historical.get("perimeter") or {} if per.get("action_id") and "SEGMENT_COUNT_CONTINUING" in cat: lines.append(f"SEGMENTS — AS FILED (before the separation): {_t('SEGMENT_COUNT_FILED', cat)} reportable segments, " f"{_t('SEGMENTS_COVERAGE', cat)} of consolidated revenue. The CONTINUING company has " f"{_t('SEGMENT_COUNT_CONTINUING', cat)} businesses. Never describe the continuing company with the " "filed count.") else: lines.append(f"SEGMENTS (filed, reconciled) — {_t('SEGMENT_COUNT_FILED', cat)} reportable segments; their revenue is " f"{_t('SEGMENTS_COVERAGE', cat)} of consolidated revenue for the same period.") elif seg.get("status") == "NOT_RECONCILED": lines.append("SEGMENTS — the filed segment data do not reconcile to consolidated revenue; no segment figures are " "available and none may be described.") dmi = s.valuation.get("damodaran_inputs") or {} if dmi.get("available"): lines.append(f"DAMODARAN INPUTS — industry {_t('DMD_INDUSTRY', cat)}; bottom-up beta {_t('DMD_BETA', cat)}; equity risk premium " f"{_t('DMD_ERP', cat)} (his implied ERP {_t('DMD_IMPLIED_ERP', cat)} for {_t('DMD_ERP_MONTH', cat)}" + (f" plus a country risk premium {_t('DMD_CRP', cat)} for {_t('DMD_COUNTRY', cat)}" if (dmi.get("crp") or 0) > 0 else "") + f"); industry sales-to-capital {_t('DMD_S2C', cat)}; industry operating margin {_t('DMD_MARGIN', cat)}.") lc = s.valuation.get("lifecycle") or {} if lc.get("stage"): import catalog as _cat _gc = _cat.comparison(s, "LIFECYCLE_TRAILING_GROWTH", "LIFECYCLE_FORWARD_GROWTH") if _gc is None or _gc[0]: lines.append(f"LIFE-CYCLE STAGE (Damodaran) — {_t('LIFECYCLE_STAGE', cat)}: trailing revenue growth {_t('LIFECYCLE_TRAILING_GROWTH', cat)}, " f"forward revenue growth {_t('LIFECYCLE_FORWARD_GROWTH', cat)} from {lc.get('forward_source')}.") else: # not comparable: never presented side by side lines.append(f"LIFE-CYCLE STAGE (Damodaran) — {_t('LIFECYCLE_STAGE', cat)}, staged on forward revenue growth " f"{_t('LIFECYCLE_FORWARD_GROWTH', cat)} ({lc.get('forward_source')}; the continuing company).") lines.append(f"TRAILING GROWTH (separate business perimeter) — trailing revenue growth {_t('LIFECYCLE_TRAILING_GROWTH', cat)} " "describes the company before the separation.") if s.ownership.get("status") == "CURRENT": lines.append(f"OWNERSHIP (SEC Form 13F, quarter ended {_t('OWN_13F_AS_OF', cat)}; filed about 45 days after quarter end. " "Permitted: state positions and changes in positions. Not permitted: inferring intent, conviction or motive, " f"or using it as a return signal) — {_t('OWN_13F_HOLDERS', cat)} institutions report a position " f"({_t('OWN_13F_HOLDERS_PREV', cat)} the quarter before; change {_t('OWN_13F_HOLDERS_CHANGE', cat)}); " f"{_t('OWN_13F_TRACKED', cat)} tracked long-term investors hold, " f"including {_t('OWN_13F_NAMES', cat)}; last quarter {_t('OWN_13F_NEW', cat)} opened, {_t('OWN_13F_ADDS', cat)} added, " f"{_t('OWN_13F_TRIMS', cat)} trimmed and {_t('OWN_13F_EXITS', cat)} exited.") sep = (s.historical.get("perimeter") or {}).get("event") if sep: lines.append(f"PERIMETER — {sep.get('business')} was separated from the company ({str(sep.get('kind') or '').replace('_', '-')}, " f"effective {sep.get('effective')}; corporate action {sep.get('action_id')}). Trailing figures include " f"{sep.get('business')}; consensus, the engine forecast and the story value the continuing company. This is the only " "corporate action that changed the perimeter.") if stry.get("ran"): lines.append(f"STORY ASSUMPTIONS (adopted storyteller of {stry.get('n_storytellers')}) — market at maturity {_t('STORY_TAM', cat)}, " f"peak share {_t('STORY_SHARE', cat)}, revenue at maturity {_t('STORY_REVENUE_MATURITY', cat)} in " f"{_t('STORY_YEARS', cat)} years, maturity margin {_t('STORY_MARGIN', cat)}; starting margin " f"{_t('STORY_START_MARGIN', cat)} ({(s.valuation.get('story_base') or {}).get('label')}); cost of capital " f"{_t('STORY_WACC', cat)} for {_t('STORY_WACC_HOLD', cat)} years" + ((f", then fading to a terminal (stable-period) rate of {_t('STORY_WACC_STABLE', cat)}" if abs(((s.valuation.get('discount_rate_path') or {}).get('terminal_wacc') or 0) - ((s.valuation.get('discount_rate_path') or {}).get('initial_wacc') or 0)) > 1e-9 else f", unchanged as the terminal (stable-period) rate {_t('STORY_WACC_STABLE', cat)}") if "STORY_WACC_STABLE" in cat else "") + f", terminal growth " f"{_t('STORY_TERMINAL_GROWTH', cat)}, failure probability {_t('STORY_FAILURE', cat)}. These are long-run " "story assumptions, not next-year forecasts." + (f" Segment build: " + "; ".join(f"{g.get('name')}" for g in stry.get("segments") or []) + "." if stry.get("segments") else "") + (" The storytellers disagree widely; treat the value as a range." if (stry.get("dispersion") or {}).get("flag") else "")) lines.append(f"MARKET PRICING — market composite {_t('MARKET_COMPOSITE', cat)} ({_t('REL_PRICE_MARKET_COMPOSITE', cat)}; " f"the weighted average of the methods was {_t('MARKET_COMPOSITE_RAW', cat)} before outlier trimming and bounds" + (f"; the mean analyst target {_t('ANALYST_TARGET', cat)} enters the composite only as a risk-adjusted " f"contribution of {_t('ANALYST_TARGET_ADJUSTED', cat)}" if "ANALYST_TARGET_ADJUSTED" in cat else "") + "); " f"forward P/E {_t('PE_FORWARD', cat)}; trailing P/E {_t('PE_TRAILING', cat)} on adjusted EPS " f"({_t('PE_TRAILING_GAAP', cat)} on reported EPS); PEG {_t('PEG', cat)}. The composite is pricing, not value.") lines.append(_peer_words(s)) lines.append(_quant_block(s)) ins = s.insiders.get("canon") or {} if ins.get("characterization"): mat = ins.get("materiality") or {} lines.append(f"INSIDERS (SEC Form 4) — [[INSIDER_CHARACTERIZATION]]. Direction of discretionary trading: " f"{ins.get('discretionary_activity_signal')}" + (f"; net discretionary trading {_t('INSIDER_DISCRETIONARY_NET', cat)}, {_t('INSIDER_MATERIALITY', cat)} of market cap " f"({'material' if mat.get('material') else 'not material'} to the valuation)" if "INSIDER_MATERIALITY" in cat else "") + f"; governance signal: {ins.get('governance_signal')}. The filings record trades, not motives: no conviction, " "confidence or intent may be inferred. Scheduled Rule 10b5-1 sales are not a governance concern.") roles = s.leadership.get("roles") or {} lines.append("LEADERSHIP — " + (", ".join(f"{r}: [[{r}]]" for r in ("CEO", "CFO") if r in roles and r in cat) or "not verified") + ". Name no other executive in these roles.") # only actions whose status a filing verifies (corporate_actions.persona_actions); recall-only actions stay in QA import corporate_actions as _ca acts = _ca.persona_actions(s.corporate_actions.get("state")) if acts: lines.append("CORPORATE ACTIONS (verified by SEC filings) — " + "; ".join( f"{a.get('business')}: {a.get('status').lower()} ({a.get('status_source')}; {a.get('action_id')})" for a in acts[:5]) + ". No other transaction status is verified; do not describe any other deal's status.") # headlines with their NewsEvidenceClaim classification: descriptive context only until a relevance model exists nclaims = ((s.evidence.get("state") or {}).get("news") or []) if nclaims: lines.append("RECENT HEADLINES (descriptive context only: never evidence of growth, moat, upside or pressure) — " + " | ".join(f"{n['title'][:110]} [{n['event_type'].lower().replace('_', ' ')}" + ("" if n["directly_about_issuer"] else "; not about the company's own fundamentals") + "]" for n in nclaims[:8])) tr = s.context.get("transcript_excerpt") if tr: lines.append("LATEST EARNINGS CALL (excerpt; qualitative context only; paraphrase, do not quote figures):\n" + tr[:3000]) return "\n".join(x for x in lines if x) def pm_anchors(s, cat): return "\n".join([ _engine_block(s), f" • price {_t('PRICE', cat)}", f" • {s.valuation.get('intrinsic_label') or 'intrinsic value'}: {_t('INTRINSIC_VALUE', cat)}. As a clause: " f"\"{_t('REL_PRICE_INTRINSIC', cat)}.\" After the subject: \"the price {_t('REL_PRICE_INTRINSIC__PREDICATE', cat)}.\"", f" • market composite (pricing): {_t('MARKET_COMPOSITE', cat)}. As a clause: \"{_t('REL_PRICE_MARKET_COMPOSITE', cat)}.\" " f"After the subject: \"the price {_t('REL_PRICE_MARKET_COMPOSITE__PREDICATE', cat)}.\"", f" • what the price requires, as a clause: \"{_t('REL_EXPECTATIONS_STORY', cat)}.\" (never after 'per', 'is' or 'shows')", f" • engine forecast EPS {_t('FORECAST_EPS', cat)}; forward P/E {_t('PE_FORWARD', cat)}; consensus EPS {_t('CONSENSUS_EPS_FY1', cat)}", _quant_block(s), ]) def evidence_offers(s): """What the PM may cite as evidence for the recommendation, and why — decided by the validated state.""" pm_rel = (s.quant.get("price_model") or {}).get("reliability") or {} pr_rel = (s.quant.get("panel_rank") or {}).get("reliability") or {} ins = s.insiders.get("canon") or {} mat = ins.get("materiality") or {} g = s.guidance.get("company") or {} stry = s.valuation.get("story") or {} iv = s.value("valuation.intrinsic_value") return [ {"id": "intrinsic_value", "description": "intrinsic value from the selected valuation engine", "eligible": iv is not None, "reason": "" if iv is not None else "no intrinsic value in the state"}, {"id": "market_pricing", "description": "market pricing composite (pricing, not value)", "eligible": s.value("pricing.market_composite") is not None, "reason": ""}, {"id": "earnings_growth", "description": "engine forecast earnings growth (reconciled forecast)", "eligible": s.value("forecast.eps_growth") is not None, "reason": ""}, {"id": "momentum", "description": "price momentum", "eligible": s.value("market.momentum.state") is not None, "reason": ""}, {"id": "consensus", "description": "sell-side consensus", "eligible": s.value("consensus.eps_fy1") is not None, "reason": ""}, {"id": "company_guidance", "description": "the company's own guidance (8-K ex-99.1)", "eligible": g.get("status") == "AVAILABLE", "reason": g.get("reason") or ""}, {"id": "macro", "description": "macro backdrop", "eligible": True, "reason": ""}, {"id": "insiders", "description": "SEC Form 4 insider transactions (trades, not motives)", "eligible": bool(mat.get("material")), "reason": ("material relative to market cap" if mat.get("material") else "discretionary insider trading is not material relative to market cap" if mat else "no insider data")}, {"id": "ownership_13f", "description": "SEC Form 13F institutional ownership (positions only, ~45-day lag)", "eligible": False, "reason": "positions, not intent; not validated as a return signal"}, {"id": "quant_price_model", "description": "single-stock quantitative price model", "eligible": bool(pm_rel.get("recommendation_evidence_allowed") not in (None, "none")), "reason": f"reliability {pm_rel.get('state')}" + (f": {pm_rel.get('basis')}" if pm_rel.get("persona_visibility") else "")}, {"id": "panel_rank", "description": "cross-sectional panel rank", "eligible": bool(pr_rel.get("recommendation_evidence_allowed") not in (None, "none")), "reason": f"reliability {pr_rel.get('state')}" + (f": {pr_rel.get('basis')}" if pr_rel.get("persona_visibility") else "")}, {"id": "story", "description": "the adopted story assumptions", "eligible": bool(stry.get("ran")), "reason": ""}, ] def panel_line(s): pr = ((s.quant.get("panel_rank") or {}).get("reliability") or {}) return f"reliability {pr.get('state')} — not usable as evidence" if not pr.get("persona_visibility") \ else f"reliability {pr.get('state')} — weak evidence only" # ── summary (after ANALYSIS_FINALIZED) ───────────────────────────────────────────────── SUMMARY_INSTRUCTIONS = """You are writing the executive summary of an equity research note. The analysis is FINAL: every number below is frozen and you may not add, change or recompute any. Write every figure ONLY as its [[TOKEN]] (list below). A sentence containing a number you typed is deleted. The rating and the blended 12-month return are [[RATING]] and [[HEADLINE_RETURN_PHRASE]]. The analyst panel's own probability-weighted scenario return is a different figure, [[PM_RETURN_PHRASE]]; never call it the headline or the blended return, and never call the headline the panel's return. Signed returns are offered only as phrases that carry their own direction: never add up/down/fell/rose/drawdown words to them. Respond with ONLY this JSON: {"summary": "<2-3 short sentences in plain English for a general reader: WHY the view is what it is. Do not restate the rating or the headline return (they are shown directly above). Use AT MOST TWO [[TOKEN]] figures. No jargon: never write 'blended', 'composite', 'mechanical', 'cross-check', 'coherence', 'engine'>", "thesis": ["<3 bullets, at most 18 words and at most ONE [[TOKEN]] figure each: the evidence for the view>"], "risks": ["<3 bullets, at most 18 words and at most ONE [[TOKEN]] figure each: what would make the view wrong>"]} The reader sees the key figures in tiles above the summary; the summary explains them, it does not list them.""" def summary_prompt(s, cat): return "\n\n".join([ SUMMARY_INSTRUCTIONS, C.persona_catalog_text(cat, s, include_deferred=True), _engine_block(s), "CONTEXT:\n" + persona_brief(s, cat)[:6000], f"PANEL — bear {_t('SCENARIO_BEAR_RETURN_PHRASE', cat)} at {_t('SCENARIO_BEAR_PROB', cat)}, neutral " f"{_t('SCENARIO_NEUTRAL_RETURN_PHRASE', cat)} at {_t('SCENARIO_NEUTRAL_PROB', cat)}, bull {_t('SCENARIO_BULL_RETURN_PHRASE', cat)} " f"at {_t('SCENARIO_BULL_PROB', cat)}; probability-weighted {_t('PM_RETURN_PHRASE', cat)}.", ]) def _parse_json(txt): if not txt: return None a, b = txt.find("{"), txt.rfind("}") if a < 0 or b <= a: return None try: return json.loads(txt[a:b + 1]) except Exception: # noqa: BLE001 return None def call_summary(prompt): """Claude first, then GPT. Returns the parsed JSON or None; failures are the caller's to report.""" key = os.getenv("ANTHROPIC_API_KEY") if key: try: import anthropic msg = anthropic.Anthropic(api_key=key).messages.create( model="claude-opus-4-7", max_tokens=1400, messages=[{"role": "user", "content": prompt}]) d = _parse_json(msg.content[0].text) if d: return d except Exception as e: # noqa: BLE001 print(f" [summary] Claude failed: {str(e)[:100]}", file=sys.stderr) key = os.getenv("OPENAI_API_KEY") if key: try: import openai r = openai.OpenAI(api_key=key).chat.completions.create( model="gpt-4o", max_tokens=1400, messages=[{"role": "system", "content": "You are a professional equity research analyst."}, {"role": "user", "content": prompt}]) return _parse_json(r.choices[0].message.content) except Exception as e: # noqa: BLE001 print(f" [summary] OpenAI failed: {str(e)[:100]}", file=sys.stderr) return None LAST_GATE_MODE = None # which gate actually answered the last call (recorded in the QA notes) def call_gate(prompt): """The semantic gate (engine/arguments.py): Claude with adaptive thinking first; if that request is rejected for its thinking configuration, Claude once without thinking; then GPT. Returns the parsed JSON ({"verdicts": [...]}) or None — the caller treats None as "gate unavailable", never as a pass. ``LAST_GATE_MODE`` records which answered.""" global LAST_GATE_MODE LAST_GATE_MODE = None key = os.getenv("ANTHROPIC_API_KEY") if key: import anthropic from llm_compat import claude_create client = anthropic.Anthropic(api_key=key, timeout=180, max_retries=1) # Claude Sonnet 5 accepts only adaptive thinking (type "enabled" with budget_tokens returns 400: live run # 18ee342c1adf2035); effort "high" is the API default. Offline, the same rubric judged with reasoning caught # 5-7 of the 10 defects the live answer-only call passed in run 2a5c13806e3f0b7f. Only text blocks are read. for mode, extra in (("claude-adaptive-thinking", {"thinking": {"type": "adaptive"}}), ("claude-no-thinking", {})): try: msg = claude_create(client, model="claude-sonnet-5", max_tokens=16000, messages=[{"role": "user", "content": prompt}], **extra) d = _parse_json("".join(getattr(b, "text", "") for b in msg.content if getattr(b, "type", "text") == "text")) if d: LAST_GATE_MODE = mode return d break except Exception as e: # noqa: BLE001 print(f" [gate] Claude ({mode}) failed: {str(e)[:300]}", file=sys.stderr) if not (extra and "thinking" in str(e).lower()): break # only a rejected thinking configuration is retried without it key = os.getenv("OPENAI_API_KEY") if key: try: import openai r = openai.OpenAI(api_key=key, timeout=120).chat.completions.create( model="gpt-4o", max_tokens=6000, temperature=0, messages=[{"role": "user", "content": prompt}]) d = _parse_json(r.choices[0].message.content) LAST_GATE_MODE = "gpt-4o" if d else None return d except Exception as e: # noqa: BLE001 print(f" [gate] OpenAI failed: {str(e)[:300]}", file=sys.stderr) return None