| """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 {}) |
| |
| 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), |
| |
| |
| 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: |
| |
| 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.") |
| |
| 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.") |
| |
| 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_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: |
| 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: |
| 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: |
| print(f" [summary] OpenAI failed: {str(e)[:100]}", file=sys.stderr) |
| return None |
|
|
|
|
| LAST_GATE_MODE = None |
|
|
|
|
| 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) |
| |
| |
| |
| 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: |
| print(f" [gate] Claude ({mode}) failed: {str(e)[:300]}", file=sys.stderr) |
| if not (extra and "thinking" in str(e).lower()): |
| break |
| 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: |
| print(f" [gate] OpenAI failed: {str(e)[:300]}", file=sys.stderr) |
| return None |
|
|
