references/exposure_framework.md
# Exposure Framework
## Overview
This document defines the scoring rules and threshold logic used by the Exposure Coach to synthesize multiple market signals into a unified exposure recommendation.
## Component Scoring (0-100 Scale)
Each input dimension is normalized to a 0-100 score where:
- **0-30**: Bearish / Risk-off signal
- **31-50**: Cautious / Neutral-bearish
- **51-70**: Neutral-bullish / Constructive
- **71-100**: Bullish / Risk-on signal
### Breadth Score
Derived from advance/decline ratios and new highs vs new lows:
| A/D Ratio (10-day) | NH/NL Ratio | Breadth Score |
|-------------------|-------------|---------------|
| > 1.5 | > 3.0 | 80-100 |
| 1.0 - 1.5 | 1.0 - 3.0 | 50-79 |
| 0.7 - 1.0 | 0.5 - 1.0 | 30-49 |
| < 0.7 | < 0.5 | 0-29 |
### Uptrend Score
Direct mapping from uptrend participation percentage:
| Uptrend % | Score |
|-----------|-------|
| > 50% | 75-100 |
| 35-50% | 50-74 |
| 20-35% | 30-49 |
| < 20% | 0-29 |
### Regime Score
Based on macro-regime-detector output:
| Regime | Score | Rationale |
|--------|-------|-----------|
| Broadening | 80 | Healthy expansion; rising tide lifts all boats |
| Concentration | 60 | Narrow leadership; selective opportunities |
| Transitional | 50 | Uncertainty; reduce commitment |
| Inflationary | 40 | Rotation stress; defensive posture |
| Contraction | 20 | Risk-off; preserve capital |
### Top Risk Score (Inverted)
Higher distribution day counts and top probability → lower score:
| Distribution Days | Top Probability | Score |
|-------------------|-----------------|-------|
| 0-2 | < 20% | 80-100 |
| 3-4 | 20-40% | 50-79 |
| 5-6 | 40-60% | 30-49 |
| 7+ | > 60% | 0-29 |
### FTD Score (Inverted)
High failure-to-deliver anomalies indicate stress:
| FTD Anomaly Level | Score |
|-------------------|-------|
| None / Low | 80-100 |
| Moderate | 50-79 |
| Elevated | 30-49 |
| Critical | 0-29 |
### Theme Score
Based on theme strength and rotation patterns:
| Theme Status | Score |
|--------------|-------|
| Strong, expanding | 80-100 |
| Stable | 50-79 |
| Rotating / Churning | 30-49 |
| Collapsing | 0-29 |
### Sector Score
Based on sector performance dispersion and leadership quality:
| Sector Condition | Score |
|------------------|-------|
| Broad strength, low dispersion | 80-100 |
| Mixed with clear leaders | 50-79 |
| High dispersion, defensive leading | 30-49 |
| Broad weakness | 0-29 |
### Institutional Score
Net institutional buying/selling trend:
| Flow Direction | Score |
|----------------|-------|
| Strong net buying | 80-100 |
| Mild net buying | 50-79 |
| Neutral / Mixed | 30-49 |
| Net selling | 0-29 |
## Composite Calculation
### Weighted Average
The composite score uses these weights:
| Component | Weight | Rationale |
|-----------|--------|-----------|
| Regime | 25% | Sets the macro baseline |
| Top Risk | 20% | Critical for downside protection |
| Breadth | 15% | Market health indicator |
| Uptrend | 15% | Participation confirmation |
| Institutional | 10% | Smart money signal |
| Sector | 5% | Rotation context |
| Theme | 5% | Thematic momentum |
| FTD | 5% | Stress indicator |
### Missing Input Handling
When inputs are missing:
1. Exclude missing components from weighted average
2. Reduce confidence level proportionally
3. Apply a 10% haircut per missing critical input (Regime, Top Risk, Breadth)
## Exposure Ceiling Mapping
| Composite Score | Exposure Ceiling |
|-----------------|------------------|
| 80-100 | 90-100% |
| 65-79 | 70-89% |
| 50-64 | 50-69% |
| 35-49 | 30-49% |
| 20-34 | 10-29% |
| 0-19 | 0-9% |
## Recommendation Logic
### NEW_ENTRY_ALLOWED
- Composite score >= 50
- Top risk score >= 40
- No critical inputs missing
### REDUCE_ONLY
- Composite score 30-49, OR
- Top risk score 25-39, OR
- 2+ critical inputs missing
### CASH_PRIORITY
- Composite score < 30, OR
- Top risk score < 25, OR
- Regime = Contraction with top risk score < 50
## Bias Determination
### Growth Bias
- Regime = Broadening or Concentration
- Theme score > 60
- Sector shows technology/growth leadership
### Value Bias
- Regime = Inflationary
- Sector shows financials/energy/materials leadership
- Institutional flow favoring value sectors
### Neutral Bias
- Conflicting signals or transitional regime
- Balanced sector performance
## Participation Assessment
### Broad Participation
- Uptrend score >= 50
- Breadth score >= 50
- Low sector dispersion
### Narrow Participation
- Uptrend score < 50
- Breadth score < 50 OR high sector dispersion
- Few sectors carrying the market
## Confidence Levels
| Condition | Confidence |
|-----------|------------|
| 6+ inputs provided, no conflicts | HIGH |
| 4-5 inputs provided OR minor conflicts | MEDIUM |
| < 4 inputs OR major conflicts | LOW |
references/regime_exposure_map.md
# Regime-to-Exposure Mapping
## Overview
This document maps macro regime states to baseline exposure ceilings and bias recommendations. The macro regime provides the foundational context; other signals adjust within these bounds.
## Regime Definitions
### Broadening
**Characteristics:**
- RSP/SPY ratio rising (equal-weight outperforming cap-weight)
- Uptrend participation > 40%
- Multiple sectors contributing to gains
- Credit spreads stable or tightening
**Baseline Exposure:** 80-100%
**Bias:** Growth-tilted
**Action:** Aggressive new entries allowed
### Concentration
**Characteristics:**
- RSP/SPY ratio falling (mega-caps dominating)
- Uptrend participation 25-40%
- Leadership in few sectors (Technology, Communications)
- Narrow breadth despite index gains
**Baseline Exposure:** 60-80%
**Bias:** Quality Growth
**Action:** Selective entries in leaders only
### Transitional
**Characteristics:**
- Cross-asset signals conflicting
- Regime indicators at boundary conditions
- High volatility in breadth measures
- Sector rotation accelerating
**Baseline Exposure:** 40-60%
**Bias:** Neutral / Defensive
**Action:** Reduce new entries; hold quality positions
### Inflationary
**Characteristics:**
- Yield curve steepening
- Commodity strength (especially energy, metals)
- Value outperforming growth
- Real assets leading
**Baseline Exposure:** 50-70%
**Bias:** Value / Real Assets
**Action:** Rotate to inflation beneficiaries
### Contraction
**Characteristics:**
- Distribution days accumulating
- Uptrend participation < 20%
- Defensive sectors (Utilities, Staples, Healthcare) leading
- Credit spreads widening
- Institutional selling detected
**Baseline Exposure:** 10-30%
**Bias:** Defensive / Cash
**Action:** CASH_PRIORITY; protect capital
## Regime Transition Signals
### Broadening → Concentration
- RSP/SPY rolling over while SPY makes new highs
- Breadth divergence (fewer stocks at highs)
- Reduce exposure ceiling by 10-20%
### Concentration → Transitional
- Mega-cap leaders showing distribution
- Volatility expansion
- Reduce exposure ceiling by 15-25%
### Transitional → Contraction
- Credit spreads widening
- Safe haven flows (bonds, gold, yen)
- Reduce exposure ceiling by 25-40%
### Contraction → Transitional
- Distribution day count resetting
- Breadth thrusts
- Begin rebuilding exposure cautiously
### Transitional → Broadening
- RSP/SPY ratio turning up
- Uptrend participation expanding
- Aggressive exposure increase warranted
## Exposure Adjustment Rules
### Within-Regime Adjustments
Even within a regime, the exposure ceiling adjusts based on:
1. **Breadth Confirmation**
- Breadth score aligning with regime: No adjustment
- Breadth diverging negatively: -10% to ceiling
- Breadth diverging positively: +5% to ceiling
2. **Top Risk Level**
- Low top risk: +10% to ceiling
- Elevated top risk: -15% to ceiling
- Critical top risk: Force CASH_PRIORITY regardless of regime
3. **Institutional Flow**
- Strong buying: +5% to ceiling
- Strong selling: -10% to ceiling
4. **FTD Anomalies**
- Critical FTD level: -15% to ceiling
- Forces extra caution on new entries
## Bias-to-Sector Mapping
| Bias | Favored Sectors | Avoid Sectors |
|------|-----------------|---------------|
| Growth | Technology, Consumer Discretionary, Communications | Utilities, Staples |
| Value | Financials, Energy, Industrials, Materials | High-multiple Growth |
| Defensive | Utilities, Healthcare, Staples | Cyclicals |
| Quality | Stable Earnings, Strong Balance Sheets | Speculative, High Debt |
## Example Scenarios
### Scenario 1: Early Bull Market
- Regime: Broadening
- Breadth: Strong (score 85)
- Top Risk: Low (score 90)
- Uptrend: Expanding (score 80)
**Result:**
- Exposure Ceiling: 95%
- Bias: Growth
- Recommendation: NEW_ENTRY_ALLOWED
### Scenario 2: Late Cycle Concentration
- Regime: Concentration
- Breadth: Weak (score 45)
- Top Risk: Moderate (score 55)
- Uptrend: Narrow (score 35)
**Result:**
- Exposure Ceiling: 55%
- Bias: Quality Growth
- Recommendation: REDUCE_ONLY
### Scenario 3: Market Top Formation
- Regime: Transitional
- Breadth: Deteriorating (score 30)
- Top Risk: High (score 25)
- Uptrend: Collapsing (score 18)
**Result:**
- Exposure Ceiling: 15%
- Bias: Defensive
- Recommendation: CASH_PRIORITY
## Integration with Other Signals
The regime provides the framework; other signals fine-tune:
1. **Theme Detector** -- Identifies which themes to favor within bias
2. **Sector Analyst** -- Confirms sector leadership alignment
3. **Institutional Flow** -- Validates smart money direction
4. **FTD Detector** -- Flags systemic stress
When signals conflict:
- Default to more conservative posture
- Reduce confidence level
- Flag conflicts in rationale
requirements.txt
# stdlib-only: packaged scripts use the Python standard library only.
scripts/calculate_exposure.py
#!/usr/bin/env python3
"""
Exposure Coach - Calculate market posture and exposure recommendation.
Synthesizes signals from multiple upstream skills to produce a unified
exposure ceiling, bias direction, and action recommendation.
"""
import argparse
import json
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
sys.path.insert(0, str(Path(__file__).resolve().parent))
from provenance_contract import validate_market_inputs
# Component weights for composite score
WEIGHTS = {
"regime": 0.25,
"top_risk": 0.20,
"breadth": 0.15,
"uptrend": 0.15,
"institutional": 0.10,
"sector": 0.05,
"theme": 0.05,
"ftd": 0.05,
}
# Critical inputs that reduce confidence when missing
CRITICAL_INPUTS = {"regime", "top_risk", "breadth"}
# Regime to baseline score mapping
REGIME_SCORES = {
"broadening": 80,
"concentration": 60,
"transitional": 50,
"inflationary": 40,
"contraction": 20,
}
def _regime_input_is_usable(data: Optional[dict]) -> bool:
"""Return whether a regime input is safe to use for exposure decisions.
Legacy flat inputs do not carry quality metadata and remain supported. For
canonical macro-regime-detector reports, explicit very-low confidence,
zero usable components, or malformed availability metadata fail closed.
"""
if not isinstance(data, dict):
return False
regime = data.get("regime")
nested_regime = isinstance(regime, dict)
if nested_regime:
if "confidence" not in regime:
return False
confidence = regime["confidence"]
if not isinstance(confidence, str):
return False
normalized = confidence.lower().strip()
if normalized not in {"high", "moderate", "low", "very_low"}:
return False
if normalized == "very_low":
return False
composite = data.get("composite")
if nested_regime and not isinstance(composite, dict):
return False
if nested_regime or (isinstance(composite, dict) and "data_quality" in composite):
if "data_quality" not in composite:
return False
data_quality = composite["data_quality"]
if (
not isinstance(data_quality, dict)
or "available_count" not in data_quality
or "total_components" not in data_quality
):
return False
available_count = data_quality["available_count"]
total_components = data_quality["total_components"]
if isinstance(available_count, bool) or not isinstance(available_count, int):
return False
if isinstance(total_components, bool) or not isinstance(total_components, int):
return False
if total_components <= 0 or not 0 < available_count <= total_components:
return False
return True
def load_json_file(path: Optional[Path]) -> Optional[dict]:
"""Load a JSON file if it exists and is valid."""
if path is None or not path.exists():
return None
try:
with open(path) as f:
return json.load(f)
except (OSError, json.JSONDecodeError) as e:
print(f"Warning: Could not load {path}: {e}", file=sys.stderr)
return None
def extract_breadth_score(data: Optional[dict]) -> Optional[int]:
"""Extract breadth score from breadth analyzer output."""
if data is None:
return None
# Support various field names from upstream skill
if "breadth_score" in data:
return int(data["breadth_score"])
if "composite_score" in data:
return int(data["composite_score"])
# market-breadth-analyzer nests its 0-100 health score under "composite"
# (100 = healthy). High = bullish, so used directly (no inversion).
composite = data.get("composite")
if isinstance(composite, dict) and "composite_score" in composite:
return int(composite["composite_score"])
if "ad_ratio" in data and "nh_nl_ratio" in data:
ad = data["ad_ratio"]
nh_nl = data["nh_nl_ratio"]
if ad > 1.5 and nh_nl > 3.0:
return 90
elif ad >= 1.0 and nh_nl >= 1.0:
return 65
elif ad >= 0.7 and nh_nl >= 0.5:
return 40
else:
return 20
return None
def _uptrend_pct_to_score(pct: float) -> int:
"""Convert an uptrend participation percentage to a 0-100 score."""
if pct > 50:
return min(100, int(50 + pct))
elif pct >= 35:
return int(35 + pct)
elif pct >= 20:
return int(20 + pct)
else:
return int(pct)
def extract_uptrend_score(data: Optional[dict]) -> Optional[int]:
"""Extract uptrend participation score.
Supports both the flat shape (``uptrend_score`` / ``uptrend_pct`` at the
top level) and the nested shape emitted by uptrend-analyzer, which stores
its result under a ``composite`` sub-object.
"""
if data is None:
return None
if "uptrend_score" in data:
return int(data["uptrend_score"])
# uptrend-analyzer nests its score under "composite"
composite = data.get("composite")
if isinstance(composite, dict):
if "composite_score" in composite:
return int(composite["composite_score"])
if "uptrend_pct" in composite:
return _uptrend_pct_to_score(composite["uptrend_pct"])
if "uptrend_pct" in data:
return _uptrend_pct_to_score(data["uptrend_pct"])
return None
def extract_regime_score(data: Optional[dict]) -> Optional[int]:
"""Extract regime score from macro-regime-detector output.
``regime`` may be a flat string (legacy) or a nested object
(``{"regime": {"current_regime": ...}}``) as emitted by
macro-regime-detector.
"""
if not _regime_input_is_usable(data):
return None
if "regime_score" in data:
return int(data["regime_score"])
regime = data.get("regime")
if isinstance(regime, dict):
name = regime.get("current_regime")
if name:
return REGIME_SCORES.get(str(name).lower().strip(), 50)
elif isinstance(regime, str):
return REGIME_SCORES.get(regime.lower().strip(), 50)
if "current_regime" in data:
return REGIME_SCORES.get(str(data["current_regime"]).lower().strip(), 50)
return None
def extract_regime_name(data: Optional[dict]) -> str:
"""Extract regime name from data.
Handles the legacy flat string form and the nested object form
(``{"regime": {"regime_label": ..., "current_regime": ...}}``). For the
nested form ``regime_label`` is preferred, falling back to
``current_regime``.
"""
if not _regime_input_is_usable(data):
return "Unknown"
regime = data.get("regime")
if isinstance(regime, dict):
label = regime.get("regime_label") or regime.get("current_regime")
return str(label).capitalize() if label else "Unknown"
if isinstance(regime, str):
return regime.capitalize()
if "current_regime" in data:
return str(data["current_regime"]).capitalize()
return "Unknown"
def extract_top_risk_score(data: Optional[dict]) -> Optional[int]:
"""Extract top risk score (inverted - high risk = low score)."""
if data is None:
return None
if "top_risk_score" in data:
return int(data["top_risk_score"])
# market-top-detector nests its 0-100 score under "composite", where HIGH =
# higher top risk (>80 = Critical/Top Formation). Invert to the
# exposure-friendly convention (high = safe to be exposed).
composite = data.get("composite")
if isinstance(composite, dict) and "composite_score" in composite:
return max(0, min(100, int(100 - composite["composite_score"])))
if "top_probability" in data:
prob = data["top_probability"]
# Invert: high probability = low score
return max(0, min(100, int(100 - prob)))
if "distribution_days" in data:
days = data["distribution_days"]
if days <= 2:
return 90
elif days <= 4:
return 65
elif days <= 6:
return 40
else:
return 15
return None
def extract_ftd_score(data: Optional[dict]) -> Optional[int]:
"""Extract Follow-Through-Day score (high = strong FTD = bullish bottom).
ftd-detector emits a 0-100 FTD quality score under ``quality_score``
(``total_score``); a confirmed Follow-Through Day is a bullish
bottom-confirmation signal, so the score is used directly (no inversion).
"""
if data is None:
return None
if "ftd_score" in data:
return int(data["ftd_score"])
# ftd-detector real shape: {"quality_score": {"total_score": 0-100, ...}}
quality = data.get("quality_score")
if isinstance(quality, dict) and quality.get("total_score") is not None:
return max(0, min(100, int(quality["total_score"])))
if "anomaly_level" in data:
level = data["anomaly_level"].lower()
mapping = {"none": 90, "low": 80, "moderate": 55, "elevated": 35, "critical": 15}
return mapping.get(level, 50)
return None
def extract_theme_score(data: Optional[dict]) -> Optional[int]:
"""Extract theme strength score."""
if data is None:
return None
if "theme_score" in data:
return int(data["theme_score"])
if "theme_strength" in data:
strength = data["theme_strength"].lower()
mapping = {"strong": 85, "stable": 65, "rotating": 40, "collapsing": 20}
return mapping.get(strength, 50)
return None
def extract_sector_score(data: Optional[dict]) -> Optional[int]:
"""Extract sector condition score."""
if data is None:
return None
if "sector_score" in data:
return int(data["sector_score"])
if "dispersion" in data and "leadership" in data:
disp = data["dispersion"]
lead = data["leadership"].lower()
if disp < 0.1 and lead in ["technology", "consumer discretionary"]:
return 85
elif disp < 0.2:
return 65
elif lead in ["utilities", "staples", "healthcare"]:
return 35
else:
return 50
return None
def extract_institutional_score(data: Optional[dict]) -> Optional[int]:
"""Extract institutional flow score."""
if data is None:
return None
if "institutional_score" in data:
return int(data["institutional_score"])
if "net_flow" in data:
flow = data["net_flow"]
if flow > 0.5:
return 90
elif flow > 0:
return 70
elif flow > -0.5:
return 40
else:
return 20
if "flow_direction" in data:
direction = data["flow_direction"].lower()
mapping = {
"strong_buying": 90,
"buying": 70,
"neutral": 50,
"selling": 30,
"strong_selling": 15,
}
return mapping.get(direction, 50)
return None
def calculate_composite_score(
scores: dict[str, Optional[int]],
) -> tuple[float, list[str], list[str]]:
"""
Calculate weighted composite score.
Returns:
Tuple of (composite_score, inputs_provided, inputs_missing)
"""
provided = []
missing = []
weighted_sum = 0.0
total_weight = 0.0
for key, weight in WEIGHTS.items():
score = scores.get(key)
if score is not None:
weighted_sum += score * weight
total_weight += weight
provided.append(key)
else:
missing.append(key)
if total_weight == 0:
return 50.0, provided, missing
composite = weighted_sum / total_weight
# Apply haircut for missing critical inputs
missing_critical = set(missing) & CRITICAL_INPUTS
haircut = len(missing_critical) * 10
composite = max(0, composite - haircut)
return composite, provided, missing
def determine_exposure_ceiling(composite: float) -> int:
"""Map composite score to exposure ceiling percentage."""
if composite >= 80:
return min(100, int(90 + (composite - 80)))
elif composite >= 65:
return int(70 + (composite - 65) * 1.3)
elif composite >= 50:
return int(50 + (composite - 50) * 1.3)
elif composite >= 35:
return int(30 + (composite - 35) * 1.3)
elif composite >= 20:
return int(10 + (composite - 20) * 1.3)
else:
return max(0, int(composite / 2))
def determine_recommendation(
composite: float, top_risk_score: Optional[int], missing_critical: int
) -> str:
"""Determine action recommendation."""
# CASH_PRIORITY conditions
if composite < 30:
return "CASH_PRIORITY"
if top_risk_score is not None and top_risk_score < 25:
return "CASH_PRIORITY"
# REDUCE_ONLY conditions
if composite < 50:
return "REDUCE_ONLY"
if top_risk_score is not None and top_risk_score < 40:
return "REDUCE_ONLY"
if missing_critical >= 2:
return "REDUCE_ONLY"
return "NEW_ENTRY_ALLOWED"
def determine_bias(
regime_name: str,
theme_score: Optional[int],
sector_data: Optional[dict],
institutional_data: Optional[dict],
) -> str:
"""Determine growth vs value bias."""
regime_lower = regime_name.lower()
# Strong regime signals
if regime_lower == "inflationary":
return "VALUE"
if regime_lower == "contraction":
return "DEFENSIVE"
# Theme strength indicates growth momentum
if theme_score is not None and theme_score > 60:
if regime_lower in ["broadening", "concentration"]:
return "GROWTH"
# Sector leadership
if sector_data and "leadership" in sector_data:
lead = sector_data["leadership"].lower()
if lead in ["technology", "consumer discretionary", "communications"]:
return "GROWTH"
if lead in ["financials", "energy", "materials", "industrials"]:
return "VALUE"
if lead in ["utilities", "staples", "healthcare"]:
return "DEFENSIVE"
# Institutional flow
if institutional_data and "sector_flows" in institutional_data:
flows = institutional_data["sector_flows"]
if isinstance(flows, dict):
growth_flow = sum(flows.get(s, 0) for s in ["Technology", "Consumer Discretionary"])
value_flow = sum(flows.get(s, 0) for s in ["Financials", "Energy", "Industrials"])
if growth_flow > value_flow + 0.2:
return "GROWTH"
if value_flow > growth_flow + 0.2:
return "VALUE"
return "NEUTRAL"
def determine_participation(
uptrend_score: Optional[int], breadth_score: Optional[int], sector_data: Optional[dict]
) -> str:
"""Assess market participation breadth."""
# Check uptrend and breadth scores
uptrend_broad = uptrend_score is not None and uptrend_score >= 50
breadth_broad = breadth_score is not None and breadth_score >= 50
# Check sector dispersion if available
low_dispersion = True
if sector_data and "dispersion" in sector_data:
low_dispersion = sector_data["dispersion"] < 0.15
if uptrend_broad and breadth_broad and low_dispersion:
return "BROAD"
elif (uptrend_broad or breadth_broad) and low_dispersion:
return "MODERATE"
else:
return "NARROW"
def determine_confidence(provided: list[str], missing: list[str]) -> str:
"""Determine confidence level based on input completeness."""
n_provided = len(provided)
missing_critical = len(set(missing) & CRITICAL_INPUTS)
if n_provided >= 6 and missing_critical == 0:
return "HIGH"
elif n_provided >= 4 or missing_critical <= 1:
return "MEDIUM"
else:
return "LOW"
def generate_rationale(
composite: float,
recommendation: str,
participation: str,
bias: str,
scores: dict[str, Optional[int]],
missing: list[str],
exposure_ceiling: Optional[int] = None,
) -> str:
"""Generate human-readable rationale.
exposure_ceiling is the value reported to the user; when omitted it is
derived from composite so the rationale never quotes the raw score.
"""
if exposure_ceiling is None:
exposure_ceiling = determine_exposure_ceiling(composite)
parts = []
# Participation assessment
if participation == "BROAD":
parts.append("Broad participation indicates healthy market internals.")
elif participation == "NARROW":
parts.append("Narrow participation suggests fragile market structure.")
# Top risk assessment
top_risk = scores.get("top_risk")
if top_risk is not None:
if top_risk >= 70:
parts.append("Low distribution day count supports risk-on posture.")
elif top_risk < 40:
parts.append("Elevated top risk signals warrant caution.")
# Regime context
regime = scores.get("regime")
if regime is not None:
if regime >= 70:
parts.append("Favorable macro regime supports elevated exposure.")
elif regime < 40:
parts.append("Challenging macro regime limits upside exposure.")
# Missing inputs
if missing:
critical_missing = sorted(set(missing) & CRITICAL_INPUTS)
if critical_missing:
parts.append(
f"Missing critical inputs ({', '.join(critical_missing)}) reduce confidence."
)
# Recommendation context
if recommendation == "CASH_PRIORITY":
parts.append("Capital preservation is the priority.")
elif recommendation == "REDUCE_ONLY":
parts.append("New entries not recommended; consider trimming on strength.")
else:
parts.append(f"New positions allowed within the {exposure_ceiling}% ceiling.")
return " ".join(parts)
def generate_markdown_report(result: dict) -> str:
"""Generate markdown report from result dict."""
lines = [
"# Market Posture Summary",
f"**Date:** {result['generated_at'][:10]} | **Confidence:** {result['confidence']}",
"",
f"## Exposure Ceiling: {result['exposure_ceiling_pct']}%",
"",
"| Dimension | Score | Status |",
"|-----------|-------|--------|",
]
# Add component scores
status_map = {
(70, 101): "Strong",
(50, 70): "Healthy",
(30, 50): "Cautious",
(0, 30): "Weak",
}
for key in [
"breadth",
"uptrend",
"regime",
"top_risk",
"ftd",
"theme",
"sector",
"institutional",
]:
score = result["component_scores"].get(f"{key}_score")
if score is not None:
status = "N/A"
for (lo, hi), label in status_map.items():
if lo <= score < hi:
status = label
break
display_name = key.replace("_", " ").title()
lines.append(f"| {display_name} | {score} | {status} |")
lines.extend(
[
"",
f"## Recommendation: {result['recommendation']}",
"",
f"**Bias:** {result['bias']}",
f"**Participation:** {result['participation']}",
"",
"### Rationale",
result["rationale"],
"",
]
)
if result["inputs_missing"]:
lines.extend(
[
"### Missing Inputs",
", ".join(result["inputs_missing"]),
"",
]
)
return "\n".join(lines)
def main():
parser = argparse.ArgumentParser(
description="Calculate market exposure posture from upstream skill outputs"
)
parser.add_argument("--breadth", type=Path, help="Path to breadth analyzer JSON")
parser.add_argument("--uptrend", type=Path, help="Path to uptrend analyzer JSON")
parser.add_argument("--regime", type=Path, help="Path to macro-regime-detector JSON")
parser.add_argument("--top-risk", type=Path, help="Path to market-top-detector JSON")
parser.add_argument("--ftd", type=Path, help="Path to ftd-detector JSON")
parser.add_argument("--theme", type=Path, help="Path to theme-detector JSON")
parser.add_argument("--sector", type=Path, help="Path to sector-analyst JSON")
parser.add_argument(
"--institutional", type=Path, help="Path to institutional-flow-tracker JSON"
)
parser.add_argument(
"--output-dir",
type=Path,
default=Path("reports"),
help="Output directory for reports (default: reports/)",
)
parser.add_argument("--json-only", action="store_true", help="Output JSON only, skip markdown")
parser.add_argument("--require-provenance", action="store_true")
parser.add_argument("--run-id")
parser.add_argument("--as-of", help="Timezone-aware evaluation timestamp")
parser.add_argument("--max-age-days", type=int, default=3)
args = parser.parse_args()
if args.require_provenance and (not args.run_id or not args.as_of):
parser.error("strict mode requires --run-id and --as-of")
# Load all inputs
breadth_data = load_json_file(args.breadth)
uptrend_data = load_json_file(args.uptrend)
regime_data = load_json_file(args.regime)
top_risk_data = load_json_file(args.top_risk)
ftd_data = load_json_file(args.ftd)
theme_data = load_json_file(args.theme)
sector_data = load_json_file(args.sector)
institutional_data = load_json_file(args.institutional)
lineage = None
if args.require_provenance:
try:
lineage = validate_market_inputs(
breadth_data,
uptrend_data,
run_id=args.run_id,
evaluation=args.as_of,
max_age_days=args.max_age_days,
)
except ValueError as exc:
parser.error(str(exc))
# Extract scores
scores: dict[str, Optional[int]] = {
"breadth": extract_breadth_score(breadth_data),
"uptrend": extract_uptrend_score(uptrend_data),
"regime": extract_regime_score(regime_data),
"top_risk": extract_top_risk_score(top_risk_data),
"ftd": extract_ftd_score(ftd_data),
"theme": extract_theme_score(theme_data),
"sector": extract_sector_score(sector_data),
"institutional": extract_institutional_score(institutional_data),
}
# Calculate composite
composite, provided, missing = calculate_composite_score(scores)
# Determine outputs
exposure_ceiling = determine_exposure_ceiling(composite)
missing_critical = len(set(missing) & CRITICAL_INPUTS)
recommendation = determine_recommendation(composite, scores["top_risk"], missing_critical)
regime_name = extract_regime_name(regime_data)
bias = determine_bias(regime_name, scores["theme"], sector_data, institutional_data)
participation = determine_participation(scores["uptrend"], scores["breadth"], sector_data)
confidence = determine_confidence(provided, missing)
rationale = generate_rationale(
composite, recommendation, participation, bias, scores, missing, exposure_ceiling
)
# Build result
now = datetime.now(timezone.utc)
result = {
"schema_version": "1.0",
"generated_at": now.isoformat(),
"exposure_ceiling_pct": exposure_ceiling,
"bias": bias,
"participation": participation,
"recommendation": recommendation,
"confidence": confidence,
"composite_score": round(composite, 1),
"component_scores": {f"{k}_score": v for k, v in scores.items() if v is not None},
"inputs_provided": provided,
"inputs_missing": missing,
"rationale": rationale,
}
if lineage is not None:
result["input_lineage"] = lineage
# Ensure output directory exists
args.output_dir.mkdir(parents=True, exist_ok=True)
# Generate timestamp for filenames
timestamp = now.strftime("%Y-%m-%d_%H%M%S")
# Write JSON
json_path = args.output_dir / f"exposure_posture_{timestamp}.json"
with open(json_path, "w", encoding="utf-8") as f:
json.dump(result, f, indent=2)
print(f"JSON report: {json_path}")
# Write markdown unless --json-only
if not args.json_only:
md_content = generate_markdown_report(result)
md_path = args.output_dir / f"exposure_posture_{timestamp}.md"
with open(md_path, "w", encoding="utf-8") as f:
f.write(md_content)
print(f"Markdown report: {md_path}")
# Print summary to stdout
print(f"\nExposure Ceiling: {exposure_ceiling}%")
print(f"Recommendation: {recommendation}")
print(f"Bias: {bias}")
print(f"Confidence: {confidence}")
return 0
if __name__ == "__main__":
sys.exit(main())
scripts/provenance_contract.py
"""Standalone common schema 1.0 and opt-in market handoff policy.
Vendored identically in the three market-regime skills. Keep schema validation
compatible with scripts/validate_provenance.py; strict policy is separate.
"""
from __future__ import annotations
from datetime import date, datetime, timedelta
from typing import Any
from urllib.parse import urlsplit, urlunsplit
from zoneinfo import ZoneInfo
SCHEMA_VERSION = "1.0"
REQUIRED_FIELDS = (
"schema_version",
"provider",
"endpoint",
"retrieved_at",
"as_of",
"timezone",
"adjusted",
"point_in_time",
"survivorship_free",
"corporate_actions_handled",
)
# Provenance fields are intentionally global so the contract is versioned in one
# place; each skill references them when emitting a block.
STRING_FIELDS = ("provider", "endpoint", "timezone", "schema_version")
BOOL_FIELDS = (
"adjusted",
"point_in_time",
"survivorship_free",
"corporate_actions_handled",
)
def _is_iso_datetime(value: Any) -> bool:
if not isinstance(value, str):
return False
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError:
return False
return parsed.tzinfo is not None
def _is_iso_date(value: Any) -> bool:
if not isinstance(value, str):
return False
try:
datetime.strptime(value, "%Y-%m-%d")
return True
except ValueError:
return False
def _is_bool(value: Any) -> bool:
return isinstance(value, bool)
def validate_provenance(block: Any, schema_version: str = SCHEMA_VERSION) -> list[str]:
"""Return a list of validation errors; an empty list means the block is valid.
Unknown fields are intentionally permitted so the contract can be extended.
Each error identifies the offending field and the requirement it violates.
"""
if not isinstance(block, dict):
return ["data_provenance must be a mapping"]
errors: list[str] = []
declared_version = block.get("schema_version")
if declared_version is not None and declared_version != schema_version:
errors.append(
f"schema_version {declared_version!r} does not match contract {schema_version!r}"
)
for field in REQUIRED_FIELDS:
if field not in block:
errors.append(f"missing required field: {field}")
for field in STRING_FIELDS:
if field in block and (not isinstance(block[field], str) or not block[field].strip()):
errors.append(f"{field} must be a non-empty string")
for field in BOOL_FIELDS:
if field in block and not _is_bool(block[field]):
errors.append(f"{field} must be a boolean (state honestly, e.g. False)")
if "retrieved_at" in block and not _is_iso_datetime(block["retrieved_at"]):
errors.append("retrieved_at must be an ISO-8601 UTC timestamp with timezone")
if "as_of" in block and not (_is_iso_date(block["as_of"]) or _is_iso_datetime(block["as_of"])):
errors.append("as_of must be an ISO-8601 date (YYYY-MM-DD)")
if "timezone" in block and isinstance(block["timezone"], str) and block["timezone"].strip():
try:
ZoneInfo(block["timezone"])
except (KeyError, ValueError):
errors.append(f"timezone {block['timezone']!r} is not a valid IANA timezone")
return errors
def build_provenance(
*,
provider: str,
endpoint: str,
retrieved_at: str,
as_of: str,
run_id: str,
timezone_name: str = "America/New_York",
) -> dict:
block = {
"schema_version": SCHEMA_VERSION,
"provider": provider,
"endpoint": endpoint,
"retrieved_at": retrieved_at,
"as_of": as_of,
"timezone": timezone_name,
"adjusted": False,
"point_in_time": False,
"survivorship_free": False,
"corporate_actions_handled": False,
"run_id": run_id,
"scope": "primary timeseries only; auxiliary source dates unverified",
}
errors = validate_provenance(block)
if errors:
raise ValueError("invalid primary provenance: " + "; ".join(errors))
if not isinstance(run_id, str) or not run_id.strip():
raise ValueError("run_id must be non-empty")
return block
def _aware(value: str, label: str) -> datetime:
if not _is_iso_datetime(value):
raise ValueError(label + " must be a timezone-aware ISO timestamp")
return datetime.fromisoformat(value.replace("Z", "+00:00"))
def validate_handoff(block: Any, *, run_id: str, evaluation: str, max_age_days: int) -> dict:
errors = validate_provenance(block)
if errors:
raise ValueError("invalid data_provenance: " + "; ".join(errors))
if not isinstance(run_id, str) or not run_id.strip() or block.get("run_id") != run_id:
raise ValueError("data_provenance run_id does not match expected run")
if isinstance(max_age_days, bool) or not isinstance(max_age_days, int) or max_age_days < 0:
raise ValueError("max_age_days must be a nonnegative integer")
now = _aware(evaluation, "evaluation")
retrieved = _aware(block["retrieved_at"], "retrieved_at")
if retrieved.utcoffset() != timedelta(0):
raise ValueError("retrieved_at must be UTC")
if retrieved > now:
raise ValueError("retrieved_at cannot be after evaluation")
value = block["as_of"]
if _is_iso_date(value):
business_date = date.fromisoformat(value)
reporting_zone = ZoneInfo(block["timezone"])
if business_date > retrieved.astimezone(reporting_zone).date():
raise ValueError("as_of cannot be after retrieval business date")
age = (now.astimezone(reporting_zone).date() - business_date).days
if age > max_age_days:
raise ValueError("primary data is stale")
else:
as_of = _aware(value, "as_of")
if as_of > retrieved:
raise ValueError("as_of cannot be after retrieval")
if (now - as_of).total_seconds() > max_age_days * 86400:
raise ValueError("primary data is stale")
return dict(block)
def validate_market_inputs(
breadth: Any, uptrend: Any, *, run_id: str, evaluation: str, max_age_days: int
) -> dict:
accepted = {}
for name, payload in (("breadth", breadth), ("uptrend", uptrend)):
if not isinstance(payload, dict):
raise ValueError(name + " input is required in strict mode")
try:
accepted[name] = validate_handoff(
payload.get("data_provenance"),
run_id=run_id,
evaluation=evaluation,
max_age_days=max_age_days,
)
except ValueError as exc:
raise ValueError(name + ": " + str(exc)) from exc
return {
"run_id": run_id,
"evaluation_at": evaluation,
"max_age_days": max_age_days,
"accepted_inputs": accepted,
"boundaries": [
"quality flags are conservative, not independently verified",
"auxiliary source dates are not verified",
"optional top-risk and other inputs are outside this gate",
"run_id is a caller consistency marker, not authentication or PIT proof",
],
}
def safe_source_url(value: str) -> str:
"""Keep source identity without credentials, query secrets or fragments."""
parsed = urlsplit(value)
host = parsed.hostname or ""
if ":" in host:
host = "[" + host + "]"
if parsed.port:
host += ":" + str(parsed.port)
return urlunsplit((parsed.scheme, host, parsed.path, "", ""))
scripts/tests/conftest.py
"""Pytest configuration for exposure-coach tests."""
import sys
from pathlib import Path
# Add scripts directory to path for imports
scripts_dir = Path(__file__).resolve().parents[1]
if str(scripts_dir) not in sys.path:
sys.path.insert(0, str(scripts_dir))
scripts/tests/test_calculate_exposure.py
"""Tests for calculate_exposure.py."""
import json
import os
import subprocess
import sys
from pathlib import Path
import pytest
from calculate_exposure import (
CRITICAL_INPUTS,
WEIGHTS,
calculate_composite_score,
determine_bias,
determine_confidence,
determine_exposure_ceiling,
determine_participation,
determine_recommendation,
extract_breadth_score,
extract_ftd_score,
extract_regime_name,
extract_regime_score,
extract_top_risk_score,
extract_uptrend_score,
generate_markdown_report,
generate_rationale,
load_json_file,
)
def canonical_regime(regime: dict) -> dict:
"""Build the availability metadata required by canonical regime reports."""
return {
"composite": {"data_quality": {"available_count": 5, "total_components": 6}},
"regime": {"confidence": "high", **regime},
}
def test_missing_critical_rationale_is_hash_seed_deterministic() -> None:
scripts_dir = Path(__file__).resolve().parents[1]
code = (
"from calculate_exposure import generate_rationale; "
"print(generate_rationale(49.0, 'REDUCE_ONLY', 'BROAD', 'NEUTRAL', "
"{'breadth': 66, 'uptrend': 72, 'regime': None, 'top_risk': None}, "
"['regime', 'top_risk']))"
)
outputs = []
for seed in ("1", "7", "31"):
env = dict(os.environ)
env["PYTHONHASHSEED"] = seed
completed = subprocess.run(
[sys.executable, "-c", code],
cwd=scripts_dir,
env=env,
check=True,
capture_output=True,
text=True,
)
outputs.append(completed.stdout.strip())
assert len(set(outputs)) == 1
assert "Missing critical inputs (regime, top_risk)" in outputs[0]
class TestExtractBreadthScore:
"""Tests for breadth score extraction."""
def test_direct_breadth_score(self):
data = {"breadth_score": 75}
assert extract_breadth_score(data) == 75
def test_composite_score_fallback(self):
data = {"composite_score": 60}
assert extract_breadth_score(data) == 60
def test_ad_ratio_calculation_high(self):
data = {"ad_ratio": 2.0, "nh_nl_ratio": 4.0}
assert extract_breadth_score(data) == 90
def test_ad_ratio_calculation_mid(self):
data = {"ad_ratio": 1.2, "nh_nl_ratio": 1.5}
assert extract_breadth_score(data) == 65
def test_ad_ratio_calculation_low(self):
data = {"ad_ratio": 0.5, "nh_nl_ratio": 0.3}
assert extract_breadth_score(data) == 20
def test_nested_composite_score(self):
# market-breadth-analyzer nests its 0-100 health score under "composite"
data = {"composite": {"composite_score": 72}}
assert extract_breadth_score(data) == 72
def test_flat_takes_priority_over_nested(self):
data = {"breadth_score": 65, "composite": {"composite_score": 10}}
assert extract_breadth_score(data) == 65
def test_non_dict_composite_ignored(self):
data = {"composite": "n/a", "ad_ratio": 2.0, "nh_nl_ratio": 4.0}
assert extract_breadth_score(data) == 90
def test_none_input(self):
assert extract_breadth_score(None) is None
def test_empty_dict(self):
assert extract_breadth_score({}) is None
class TestExtractUptrendScore:
"""Tests for uptrend score extraction."""
def test_direct_score(self):
data = {"uptrend_score": 80}
assert extract_uptrend_score(data) == 80
def test_uptrend_pct_high(self):
data = {"uptrend_pct": 60}
score = extract_uptrend_score(data)
assert score >= 75
def test_uptrend_pct_mid(self):
data = {"uptrend_pct": 40}
score = extract_uptrend_score(data)
assert 50 <= score <= 80
def test_uptrend_pct_low(self):
data = {"uptrend_pct": 15}
score = extract_uptrend_score(data)
assert score < 30
def test_nested_composite_score(self):
# uptrend-analyzer stores its score under "composite"
data = {"composite": {"composite_score": 72}}
assert extract_uptrend_score(data) == 72
def test_nested_composite_uptrend_pct(self):
data = {"composite": {"uptrend_pct": 60}}
assert extract_uptrend_score(data) >= 75
def test_flat_takes_priority_over_nested(self):
data = {"uptrend_score": 80, "composite": {"composite_score": 10}}
assert extract_uptrend_score(data) == 80
def test_non_dict_composite_ignored(self):
data = {"composite": "n/a", "uptrend_pct": 40}
score = extract_uptrend_score(data)
assert 50 <= score <= 80
def test_none_input(self):
assert extract_uptrend_score(None) is None
def test_empty_dict(self):
assert extract_uptrend_score({}) is None
class TestExtractRegimeScore:
"""Tests for regime score extraction."""
def test_broadening_regime(self):
data = {"regime": "Broadening"}
assert extract_regime_score(data) == 80
def test_contraction_regime(self):
data = {"regime": "contraction"}
assert extract_regime_score(data) == 20
def test_current_regime_field(self):
data = {"current_regime": "Transitional"}
assert extract_regime_score(data) == 50
def test_direct_regime_score(self):
data = {"regime_score": 65}
assert extract_regime_score(data) == 65
def test_nested_regime_dict_current_regime(self):
# macro-regime-detector emits regime as a nested object
data = canonical_regime({"current_regime": "Broadening"})
assert extract_regime_score(data) == 80
def test_nested_regime_dict_unknown_defaults_50(self):
data = canonical_regime({"current_regime": "Sideways"})
assert extract_regime_score(data) == 50
def test_nested_regime_dict_no_current_regime(self):
data = canonical_regime({"regime_label": "Risk-On"})
assert extract_regime_score(data) is None
def test_canonical_very_low_confidence_is_missing(self):
data = {
"composite": {"data_quality": {"available_count": 5, "total_components": 6}},
"regime": {"current_regime": "Broadening", "confidence": "very_low"},
}
assert extract_regime_score(data) is None
assert extract_regime_name(data) == "Unknown"
def test_canonical_zero_available_components_is_missing(self):
data = {
"composite": {"data_quality": {"available_count": 0, "total_components": 6}},
"regime": {"current_regime": "Concentration", "confidence": "low"},
}
assert extract_regime_score(data) is None
assert extract_regime_name(data) == "Unknown"
def test_canonical_malformed_availability_is_missing(self):
data = {
"composite": {"data_quality": {"available_count": "six", "total_components": 6}},
"regime": {"current_regime": "Broadening", "confidence": "high"},
}
assert extract_regime_score(data) is None
@pytest.mark.parametrize(
"data",
[
{
"regime": {"current_regime": "Broadening", "confidence": "high"},
},
{
"composite": "not-an-object",
"regime": {"current_regime": "Broadening", "confidence": "high"},
},
{
"composite": {},
"regime": {"current_regime": "Broadening", "confidence": "high"},
},
{
"composite": {"data_quality": {"available_count": 5, "total_components": 6}},
"regime": {"current_regime": "Broadening"},
},
],
ids=[
"missing-composite",
"non-object-composite",
"missing-data-quality",
"missing-confidence",
],
)
def test_incomplete_canonical_availability_is_missing(self, data):
assert extract_regime_score(data) is None
assert extract_regime_name(data) == "Unknown"
@pytest.mark.parametrize("available_count", [float("nan"), float("inf"), 0.5, 7, True])
def test_canonical_adversarial_availability_is_missing(self, available_count):
data = {
"composite": {
"data_quality": {"available_count": available_count, "total_components": 6}
},
"regime": {"current_regime": "Broadening", "confidence": "high"},
}
assert extract_regime_score(data) is None
assert extract_regime_name(data) == "Unknown"
@pytest.mark.parametrize("data", [[], "regime", 42, True])
def test_non_object_regime_input_is_missing(self, data):
assert extract_regime_score(data) is None
assert extract_regime_name(data) == "Unknown"
def test_canonical_missing_or_invalid_total_is_missing(self):
missing_total = {
"composite": {"data_quality": {"available_count": 5}},
"regime": {"current_regime": "Broadening", "confidence": "high"},
}
invalid_total = {
"composite": {"data_quality": {"available_count": 5, "total_components": False}},
"regime": {"current_regime": "Broadening", "confidence": "high"},
}
assert extract_regime_score(missing_total) is None
assert extract_regime_score(invalid_total) is None
def test_canonical_low_confidence_with_data_is_accepted(self):
data = {
"composite": {"data_quality": {"available_count": 5, "total_components": 6}},
"regime": {"current_regime": "Broadening", "confidence": "low"},
}
assert extract_regime_score(data) == 80
assert extract_regime_name(data) == "Broadening"
def test_none_input(self):
assert extract_regime_score(None) is None
def test_empty_dict(self):
assert extract_regime_score({}) is None
class TestExtractRegimeName:
"""Tests for regime name extraction (incl. nested dict regression)."""
def test_flat_string_regime(self):
assert extract_regime_name({"regime": "broadening"}) == "Broadening"
def test_flat_current_regime(self):
assert extract_regime_name({"current_regime": "contraction"}) == "Contraction"
def test_nested_label_preferred(self):
data = canonical_regime({"regime_label": "Risk-On", "current_regime": "broadening"})
assert extract_regime_name(data) == "Risk-on"
def test_nested_current_regime_fallback(self):
data = canonical_regime({"current_regime": "transitional"})
assert extract_regime_name(data) == "Transitional"
def test_nested_empty_dict_returns_unknown(self):
assert extract_regime_name({"regime": {}}) == "Unknown"
def test_dict_input_does_not_raise(self):
# Regression: previously data["regime"].capitalize() raised on dict
data = {"regime": {"current_regime": "broadening"}}
result = extract_regime_name(data)
assert isinstance(result, str)
def test_none_input(self):
assert extract_regime_name(None) == "Unknown"
def test_empty_dict(self):
assert extract_regime_name({}) == "Unknown"
class TestExtractTopRiskScore:
"""Tests for top risk score extraction."""
def test_direct_score(self):
data = {"top_risk_score": 30}
assert extract_top_risk_score(data) == 30
def test_top_probability_high(self):
# High probability = low score (inverted)
data = {"top_probability": 80}
assert extract_top_risk_score(data) == 20
def test_top_probability_low(self):
# Low probability = high score
data = {"top_probability": 10}
assert extract_top_risk_score(data) == 90
def test_distribution_days_few(self):
data = {"distribution_days": 1}
assert extract_top_risk_score(data) == 90
def test_distribution_days_many(self):
data = {"distribution_days": 8}
assert extract_top_risk_score(data) == 15
def test_nested_composite_inverted_high_risk(self):
# market-top-detector composite=85 (Critical/Top Formation) -> low score
data = {"composite": {"composite_score": 85}}
assert extract_top_risk_score(data) == 15
def test_nested_composite_inverted_low_risk(self):
# composite=15 (Green/Normal) -> high (safe) score
data = {"composite": {"composite_score": 15}}
assert extract_top_risk_score(data) == 85
def test_flat_takes_priority_over_nested(self):
# explicit top_risk_score is already exposure-friendly; not inverted
data = {"top_risk_score": 40, "composite": {"composite_score": 85}}
assert extract_top_risk_score(data) == 40
class TestExtractFtdScore:
"""Tests for Follow-Through-Day score extraction (high = bullish, NOT inverted)."""
def test_direct_ftd_score(self):
assert extract_ftd_score({"ftd_score": 70}) == 70
def test_nested_quality_score_strong(self):
# ftd-detector real shape: strong FTD -> high score (bullish, direct)
data = {"quality_score": {"total_score": 82, "signal": "Strong FTD"}}
assert extract_ftd_score(data) == 82
def test_nested_quality_score_no_ftd(self):
data = {"quality_score": {"total_score": 0, "signal": "No FTD"}}
assert extract_ftd_score(data) == 0
def test_legacy_anomaly_level_still_supported(self):
assert extract_ftd_score({"anomaly_level": "none"}) == 90
def test_none_and_empty(self):
assert extract_ftd_score(None) is None
assert extract_ftd_score({}) is None
class TestRealUpstreamShapesAllCount:
"""Regression: the real upstream JSON shapes must all produce a score.
Reproduces the reported bug where breadth/top_risk/ftd silently returned
None (only regime + uptrend counted), forcing a missing-critical haircut
and a CASH_PRIORITY / LOW-confidence verdict.
"""
def test_all_five_inputs_extracted(self):
breadth = {"composite": {"composite_score": 70}} # market-breadth-analyzer
uptrend = {"composite": {"composite_score": 65}} # uptrend-analyzer
regime = canonical_regime({"current_regime": "broadening"})
top_risk = {"composite": {"composite_score": 20}} # market-top-detector (low risk)
ftd = {"quality_score": {"total_score": 75}} # ftd-detector (strong FTD)
scores = {
"breadth": extract_breadth_score(breadth),
"uptrend": extract_uptrend_score(uptrend),
"regime": extract_regime_score(regime),
"top_risk": extract_top_risk_score(top_risk),
"ftd": extract_ftd_score(ftd),
}
# The bug: breadth/top_risk/ftd were None. All five must now resolve.
assert all(v is not None for v in scores.values()), scores
assert scores["breadth"] == 70
assert scores["top_risk"] == 80 # inverted: 100 - 20
assert scores["ftd"] == 75 # direct
composite, provided, missing = calculate_composite_score(
{**scores, "institutional": None, "sector": None, "theme": None}
)
# No critical input missing -> no haircut; healthy composite, not cash-priority
assert set(missing).isdisjoint(CRITICAL_INPUTS)
assert composite > 50
class TestCalculateCompositeScore:
"""Tests for composite score calculation."""
def test_all_inputs_provided(self):
scores = {
"regime": 80,
"top_risk": 70,
"breadth": 65,
"uptrend": 60,
"institutional": 75,
"sector": 70,
"theme": 65,
"ftd": 80,
}
composite, provided, missing = calculate_composite_score(scores)
assert len(provided) == 8
assert len(missing) == 0
# Weighted average check
expected = sum(scores[k] * WEIGHTS[k] for k in WEIGHTS)
assert abs(composite - expected) < 0.1
def test_missing_critical_inputs(self):
scores = {
"regime": None, # critical
"top_risk": None, # critical
"breadth": 65, # critical but present
"uptrend": 60,
"institutional": 75,
"sector": 70,
"theme": 65,
"ftd": 80,
}
composite, provided, missing = calculate_composite_score(scores)
assert "regime" in missing
assert "top_risk" in missing
# Haircut applied: 2 critical missing * 10 = 20
assert len(provided) == 6
def test_no_inputs(self):
scores = {k: None for k in WEIGHTS}
composite, provided, missing = calculate_composite_score(scores)
assert composite == 50.0 # Default when no inputs
assert len(provided) == 0
assert len(missing) == 8
class TestDetermineExposureCeiling:
"""Tests for exposure ceiling mapping."""
def test_high_composite(self):
assert determine_exposure_ceiling(90) >= 90
def test_mid_composite(self):
ceiling = determine_exposure_ceiling(60)
assert 50 <= ceiling <= 80
def test_low_composite(self):
ceiling = determine_exposure_ceiling(25)
assert ceiling <= 30
def test_very_low_composite(self):
ceiling = determine_exposure_ceiling(10)
assert ceiling <= 10
class TestDetermineRecommendation:
"""Tests for recommendation logic."""
def test_cash_priority_low_composite(self):
rec = determine_recommendation(25, 50, 0)
assert rec == "CASH_PRIORITY"
def test_cash_priority_low_top_risk(self):
rec = determine_recommendation(60, 20, 0)
assert rec == "CASH_PRIORITY"
def test_reduce_only_mid_composite(self):
rec = determine_recommendation(45, 50, 0)
assert rec == "REDUCE_ONLY"
def test_reduce_only_missing_critical(self):
rec = determine_recommendation(60, 50, 2)
assert rec == "REDUCE_ONLY"
def test_new_entry_allowed(self):
rec = determine_recommendation(70, 60, 0)
assert rec == "NEW_ENTRY_ALLOWED"
class TestDetermineBias:
"""Tests for bias determination."""
def test_inflationary_regime(self):
bias = determine_bias("Inflationary", 50, None, None)
assert bias == "VALUE"
def test_contraction_regime(self):
bias = determine_bias("Contraction", 50, None, None)
assert bias == "DEFENSIVE"
def test_broadening_with_strong_theme(self):
bias = determine_bias("Broadening", 75, None, None)
assert bias == "GROWTH"
def test_sector_leadership_technology(self):
sector_data = {"leadership": "Technology"}
bias = determine_bias("Transitional", 50, sector_data, None)
assert bias == "GROWTH"
def test_sector_leadership_financials(self):
sector_data = {"leadership": "Financials"}
bias = determine_bias("Transitional", 50, sector_data, None)
assert bias == "VALUE"
def test_neutral_default(self):
bias = determine_bias("Transitional", 50, None, None)
assert bias == "NEUTRAL"
class TestDetermineParticipation:
"""Tests for participation assessment."""
def test_broad_participation(self):
part = determine_participation(70, 65, {"dispersion": 0.05})
assert part == "BROAD"
def test_narrow_participation(self):
part = determine_participation(30, 35, {"dispersion": 0.25})
assert part == "NARROW"
def test_moderate_participation(self):
part = determine_participation(55, 40, {"dispersion": 0.10})
assert part == "MODERATE"
class TestDetermineConfidence:
"""Tests for confidence level."""
def test_high_confidence(self):
provided = list(WEIGHTS.keys())[:6]
missing = list(WEIGHTS.keys())[6:]
# Remove critical from missing
missing = [m for m in missing if m not in CRITICAL_INPUTS]
conf = determine_confidence(provided, missing)
assert conf == "HIGH"
def test_medium_confidence(self):
provided = ["regime", "breadth", "uptrend", "sector"]
missing = ["top_risk", "ftd", "theme", "institutional"]
conf = determine_confidence(provided, missing)
assert conf == "MEDIUM"
def test_low_confidence(self):
provided = ["sector", "theme"]
missing = ["regime", "top_risk", "breadth", "uptrend", "ftd", "institutional"]
conf = determine_confidence(provided, missing)
assert conf == "LOW"
class TestGenerateRationale:
"""Tests for rationale generation."""
def test_rationale_includes_participation(self):
rationale = generate_rationale(
70, "NEW_ENTRY_ALLOWED", "BROAD", "GROWTH", {"top_risk": 80, "regime": 75}, []
)
assert "Broad participation" in rationale
def test_rationale_includes_missing_inputs(self):
rationale = generate_rationale(
60, "REDUCE_ONLY", "MODERATE", "NEUTRAL", {"breadth": 60}, ["regime", "top_risk"]
)
assert "Missing critical inputs" in rationale
def test_rationale_cash_priority(self):
rationale = generate_rationale(
25, "CASH_PRIORITY", "NARROW", "DEFENSIVE", {"top_risk": 20}, []
)
assert "preservation" in rationale.lower()
def test_rationale_quotes_exposure_ceiling_not_composite(self):
"""The ceiling sentence must report the ceiling, never the raw score."""
rationale = generate_rationale(
73.1,
"NEW_ENTRY_ALLOWED",
"BROAD",
"NEUTRAL",
{"breadth": 79, "uptrend": 57},
[],
exposure_ceiling=80,
)
assert "80% ceiling" in rationale
assert "73% ceiling" not in rationale
def test_rationale_ceiling_defaults_to_derived_value(self):
"""Omitting exposure_ceiling still yields the ceiling, not the composite."""
composite = 73.1
expected = determine_exposure_ceiling(composite)
rationale = generate_rationale(
composite, "NEW_ENTRY_ALLOWED", "BROAD", "NEUTRAL", {"breadth": 79}, []
)
assert f"{expected}% ceiling" in rationale
assert f"{int(composite)}% ceiling" not in rationale
class TestGenerateMarkdownReport:
"""Tests for markdown report generation."""
def test_markdown_contains_exposure(self):
result = {
"generated_at": "2026-03-16T07:00:00Z",
"confidence": "HIGH",
"exposure_ceiling_pct": 75,
"component_scores": {
"breadth_score": 65,
"regime_score": 80,
},
"recommendation": "NEW_ENTRY_ALLOWED",
"bias": "GROWTH",
"participation": "BROAD",
"rationale": "Test rationale.",
"inputs_missing": [],
}
md = generate_markdown_report(result)
assert "75%" in md
assert "NEW_ENTRY_ALLOWED" in md
assert "GROWTH" in md
def test_markdown_includes_missing(self):
result = {
"generated_at": "2026-03-16T07:00:00Z",
"confidence": "MEDIUM",
"exposure_ceiling_pct": 50,
"component_scores": {"breadth_score": 60},
"recommendation": "REDUCE_ONLY",
"bias": "NEUTRAL",
"participation": "NARROW",
"rationale": "Caution advised.",
"inputs_missing": ["regime", "top_risk"],
}
md = generate_markdown_report(result)
assert "Missing Inputs" in md
assert "regime" in md
class TestLoadJsonFile:
"""Tests for JSON file loading."""
def test_load_valid_file(self, tmp_path):
test_file = tmp_path / "test.json"
test_data = {"key": "value"}
test_file.write_text(json.dumps(test_data), encoding="utf-8")
result = load_json_file(test_file)
assert result == test_data
def test_load_nonexistent_file(self, tmp_path):
result = load_json_file(tmp_path / "nonexistent.json")
assert result is None
def test_load_none_path(self):
result = load_json_file(None)
assert result is None
def test_load_invalid_json(self, tmp_path):
test_file = tmp_path / "invalid.json"
test_file.write_text("not valid json", encoding="utf-8")
result = load_json_file(test_file)
assert result is None
class TestIntegration:
"""Integration tests for the full pipeline."""
def test_full_pipeline_with_all_inputs(self, tmp_path):
"""Test complete flow with all inputs provided."""
import sys
from calculate_exposure import main
# Create mock input files
breadth_file = tmp_path / "breadth.json"
breadth_file.write_text(json.dumps({"breadth_score": 70}), encoding="utf-8")
regime_file = tmp_path / "regime.json"
regime_file.write_text(json.dumps({"regime": "Broadening"}), encoding="utf-8")
top_risk_file = tmp_path / "top_risk.json"
top_risk_file.write_text(json.dumps({"top_risk_score": 75}), encoding="utf-8")
uptrend_file = tmp_path / "uptrend.json"
uptrend_file.write_text(json.dumps({"uptrend_score": 65}), encoding="utf-8")
output_dir = tmp_path / "reports"
# Mock sys.argv
original_argv = sys.argv
sys.argv = [
"calculate_exposure.py",
"--breadth",
str(breadth_file),
"--regime",
str(regime_file),
"--top-risk",
str(top_risk_file),
"--uptrend",
str(uptrend_file),
"--output-dir",
str(output_dir),
"--json-only",
]
try:
result = main()
assert result == 0
# Check output files exist
json_files = list(output_dir.glob("exposure_posture_*.json"))
assert len(json_files) == 1
# Validate JSON content
with open(json_files[0]) as f:
data = json.load(f)
assert "exposure_ceiling_pct" in data
assert "recommendation" in data
assert data["confidence"] in ["HIGH", "MEDIUM", "LOW"]
finally:
sys.argv = original_argv
def test_partial_inputs_reduce_confidence(self, tmp_path):
"""Test that missing critical inputs reduce confidence."""
import sys
from calculate_exposure import main
# Create only one non-critical input
sector_file = tmp_path / "sector.json"
sector_file.write_text(json.dumps({"sector_score": 60}), encoding="utf-8")
output_dir = tmp_path / "reports"
original_argv = sys.argv
sys.argv = [
"calculate_exposure.py",
"--sector",
str(sector_file),
"--output-dir",
str(output_dir),
"--json-only",
]
try:
result = main()
assert result == 0
json_files = list(output_dir.glob("exposure_posture_*.json"))
with open(json_files[0]) as f:
data = json.load(f)
# All critical inputs missing → LOW confidence
assert data["confidence"] == "LOW"
# Missing critical inputs triggers haircut → lower exposure
assert data["exposure_ceiling_pct"] < 50
finally:
sys.argv = original_argv
def test_unusable_regime_matches_missing_input_safety_baseline(self, tmp_path):
"""Issue #311: a 0/6 report cannot raise exposure or confidence."""
import sys
from calculate_exposure import main
breadth_file = tmp_path / "breadth.json"
breadth_file.write_text(json.dumps({"breadth_score": 76}), encoding="utf-8")
uptrend_file = tmp_path / "uptrend.json"
uptrend_file.write_text(json.dumps({"uptrend_score": 59}), encoding="utf-8")
regime_file = tmp_path / "regime.json"
regime_file.write_text(
json.dumps(
{
"composite": {
"composite_score": 0.0,
"signaling_components": 0,
"data_quality": {"available_count": 0, "total_components": 6},
},
"regime": {
"current_regime": "concentration",
"confidence": "very_low",
},
}
),
encoding="utf-8",
)
output_dir = tmp_path / "reports"
original_argv = sys.argv
sys.argv = [
"calculate_exposure.py",
"--breadth",
str(breadth_file),
"--uptrend",
str(uptrend_file),
"--regime",
str(regime_file),
"--output-dir",
str(output_dir),
"--json-only",
]
try:
assert main() == 0
finally:
sys.argv = original_argv
result = json.loads(
next(output_dir.glob("exposure_posture_*.json")).read_text(encoding="utf-8")
)
assert result["component_scores"] == {
"breadth_score": 76,
"uptrend_score": 59,
}
assert result["inputs_provided"] == ["breadth", "uptrend"]
assert "regime" in result["inputs_missing"]
assert result["confidence"] == "LOW"
assert result["composite_score"] == 47.5
assert result["exposure_ceiling_pct"] == 46
assert result["bias"] == "NEUTRAL"
scripts/tests/test_provenance_contract.py
"""Standalone copies retain the common schema while strict policy is opt-in."""
import importlib.util
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[4]
def load(name, path):
spec = importlib.util.spec_from_file_location(name, path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
contract = load("market_contract", Path(__file__).resolve().parents[1] / "provenance_contract.py")
shared = load("shared_provenance", ROOT / "scripts/validate_provenance.py")
def block(**overrides):
result = contract.build_provenance(
provider="fixture",
endpoint="fixture://primary",
retrieved_at="2026-10-17T01:00:00Z",
as_of="2026-10-16",
run_id="run-a",
)
result.update(overrides)
return result
def validate(value, **overrides):
args = dict(run_id="run-a", evaluation="2026-10-17T02:00:00Z", max_age_days=0)
args.update(overrides)
return contract.validate_handoff(value, **args)
def test_vendored_helpers_are_identical():
copies = [
(ROOT / "skills" / skill / "scripts/provenance_contract.py").read_bytes()
for skill in ("market-breadth-analyzer", "uptrend-analyzer", "exposure-coach")
]
assert copies[0] == copies[1] == copies[2]
@pytest.mark.parametrize(
"changes",
[
{},
{"schema_version": "0.9"},
{"adjusted": 1},
{"as_of": None},
{"retrieved_at": "2026-10-17T02:00:00+01:00"},
{"timezone": "invalid"},
],
)
def test_common_schema_matches_shared_validator(changes):
value = block(**changes)
assert contract.validate_provenance(value) == shared.validate_provenance(value)
def test_date_age_uses_reporting_timezone():
# Both instants are October 16 in New York, so this is zero business days old.
assert validate(block())["as_of"] == "2026-10-16"
with pytest.raises(ValueError, match="stale"):
validate(block(), evaluation="2026-10-17T05:00:00Z")
assert validate(block(), evaluation="2026-10-17T05:00:00Z", max_age_days=1)
def test_instant_age_boundary_is_elapsed_time():
value = block(as_of="2026-10-16T01:00:00Z")
assert validate(value, evaluation="2026-10-17T01:00:00Z", max_age_days=1)
with pytest.raises(ValueError, match="stale"):
validate(value, evaluation="2026-10-17T01:00:01Z", max_age_days=1)
@pytest.mark.parametrize(
"changes, message",
[
({"run_id": "other-run"}, "run_id"),
({"schema_version": "2.0"}, "schema_version"),
({"retrieved_at": "2026-10-17T03:00:00Z"}, "after evaluation"),
({"retrieved_at": "2026-10-17T02:00:00+01:00"}, "UTC"),
({"retrieved_at": "2026-10-17T01:00:00"}, "timezone"),
({"as_of": "2026-10-17"}, "after retrieval business date"),
({"as_of": "2026-10-17T01:00:01Z"}, "after retrieval"),
],
)
def test_strict_rejections(changes, message):
with pytest.raises(ValueError, match=message):
validate(block(**changes))
@pytest.mark.parametrize("age", [-1, True, 0.5])
def test_invalid_age_policy_rejected(age):
with pytest.raises(ValueError, match="nonnegative integer"):
validate(block(), max_age_days=age)
def test_both_primary_inputs_required():
with pytest.raises(ValueError, match="uptrend input is required"):
contract.validate_market_inputs(
{"data_provenance": block()},
None,
run_id="run-a",
evaluation="2026-10-17T02:00:00Z",
max_age_days=0,
)
@pytest.mark.parametrize("value", [None, "wrong", [], 3])
def test_schema_nonmapping_rejected(value):
assert contract.validate_provenance(value) == shared.validate_provenance(value)
def test_schema_missing_and_invalid_fields_match():
assert contract.validate_provenance({}) == shared.validate_provenance({})
for field in contract.STRING_FIELDS:
for value in (None, "", 3):
candidate = block(**{field: value})
assert contract.validate_provenance(candidate) == shared.validate_provenance(candidate)
for value in (None, 3, "bogus", "2026-02-30"):
assert contract.validate_provenance(block(as_of=value))
for value in (None, 3, "bogus", "2026-10-17T01:00:00"):
assert contract.validate_provenance(block(retrieved_at=value))
def test_build_rejects_invalid_date_and_empty_run():
for changes in ({"as_of": "invalid"}, {"run_id": ""}):
args = dict(
provider="fixture",
endpoint="fixture://primary",
retrieved_at="2026-10-17T01:00:00Z",
as_of="2026-10-16",
run_id="run-a",
)
args.update(changes)
with pytest.raises(ValueError):
contract.build_provenance(**args)
def test_invalid_evaluation_and_wrapped_input_error():
with pytest.raises(ValueError, match="evaluation"):
validate(block(), evaluation="invalid")
with pytest.raises(ValueError, match="breadth:"):
contract.validate_market_inputs(
{"data_provenance": {}},
{"data_provenance": block()},
run_id="run-a",
evaluation="2026-10-17T02:00:00Z",
max_age_days=0,
)
lineage = contract.validate_market_inputs(
{"data_provenance": block()},
{"data_provenance": block()},
run_id="run-a",
evaluation="2026-10-17T02:00:00Z",
max_age_days=0,
)
assert set(lineage["accepted_inputs"]) == {"breadth", "uptrend"}
assert lineage["boundaries"]
@pytest.mark.parametrize(
"url, expected",
[
(
"https://user:secret@example.com:8443/data.csv?token=secret#secret", # pragma: allowlist secret
"https://example.com:8443/data.csv",
),
("https://[::1]:8443/data.csv?token=secret", "https://[::1]:8443/data.csv"),
("/local.csv?token=secret", "/local.csv"),
],
)
def test_source_identity_strips_credentials_and_query(url, expected):
assert contract.safe_source_url(url) == expected
SKILL.md
---
name: exposure-coach
description: Generate a one-page Market Posture summary with net exposure ceiling, growth-vs-value bias, participation breadth, and new-entry-allowed vs cash-priority recommendation by integrating signals from breadth, regime, and flow analysis skills.
---
# Exposure Coach
## Overview
Exposure Coach synthesizes outputs from market-breadth-analyzer, uptrend-analyzer, macro-regime-detector, market-top-detector, ftd-detector, theme-detector, sector-analyst, and institutional-flow-tracker into a unified control-plane decision. The skill answers the solo trader's core question: "How much capital should I commit to equities right now?" before any individual stock analysis begins.
## When to Use
- Before initiating any new stock positions to determine appropriate capital commitment
- At the start of each trading week to calibrate portfolio exposure
- When multiple market signals conflict and a unified posture is needed
- After significant macro or market events to reassess exposure ceiling
- When transitioning between market regimes (broadening, concentration, contraction)
## Prerequisites
- Python 3.9+
- FMP API key (set `FMP_API_KEY` environment variable) for institutional-flow-tracker data
- Input JSON files from upstream skills (see Workflow Step 1)
- Standard library + `argparse`, `json`, `datetime`
## Workflow
### Step 1: Gather Upstream Skill Outputs
Collect the most recent JSON outputs from integrated skills. Each file provides a specific signal dimension:
| Skill | Output File Pattern | Signal Provided |
|-------|---------------------|-----------------|
| market-breadth-analyzer | `breadth_*.json` | Advance/decline ratios, new highs/lows |
| uptrend-analyzer | `uptrend_*.json` | Uptrend participation percentage |
| macro-regime-detector | `regime_*.json` | Current regime (Concentration, Broadening, etc.) |
| market-top-detector | `top_risk_*.json` | Distribution day count, top probability score |
| ftd-detector | `ftd_*.json` | Follow-Through Day quality (market bottom confirmation) |
| theme-detector | `theme_detector_*.json` or `theme_*.json` | Active investment themes and rotation |
| sector-analyst | `sector_*.json` | Sector performance rankings |
| institutional-flow-tracker | `institutional_*.json` | Net institutional buying/selling |
### Step 2: Run Exposure Scoring Engine
Execute the exposure scoring script with paths to upstream outputs:
```bash
python3 skills/exposure-coach/scripts/calculate_exposure.py \
--breadth reports/breadth_latest.json \
--uptrend reports/uptrend_latest.json \
--regime reports/regime_latest.json \
--top-risk reports/top_risk_latest.json \
--ftd reports/ftd_latest.json \
--theme reports/theme_latest.json \
--sector reports/sector_latest.json \
--institutional reports/institutional_latest.json \
--output-dir reports/
```
The script accepts partial inputs; missing files reduce confidence but do not block execution.
Canonical macro-regime reports must include nested `regime.confidence` and
`composite.data_quality` with valid integer component counts. Missing or malformed
availability metadata, `very_low` confidence, and zero usable components are
treated as missing critical input. They do not contribute a regime score or bias,
and the normal missing-input haircut and confidence cap apply. Never override this
degradation by manually copying the report's regime label into the exposure decision.
**Verification pitfall:** After each run, inspect the generated JSON fields `inputs_provided` and `inputs_missing`. If a file you passed on the CLI still appears in `inputs_missing` (for example a theme-detector JSON that the exposure engine did not recognize), report the affected dimension as degraded and keep confidence capped; do not assume the supplied input was incorporated just because the CLI argument was present.
**Theme-detector ingestion caveat:** The theme detector commonly emits `theme_detector_YYYY-MM-DD_HHMMSS.json` with a `themes` object. If that file is not recognized by `calculate_exposure.py` and `theme` remains in `inputs_missing`, do not fold theme strength into the exposure ceiling manually. Instead, keep the Exposure Coach confidence capped, state that the theme dimension was not incorporated, and summarize theme/sector findings separately in the broader trading brief.
### Step 3: Interpret the Market Posture Summary
Review the generated posture report containing:
1. **Exposure Ceiling** -- Maximum recommended equity allocation (0-100%)
2. **Bias Direction** -- Growth vs Value tilt based on regime and flow
3. **Participation Assessment** -- Broad (healthy) vs Narrow (fragile) market
4. **Action Recommendation** -- NEW_ENTRY_ALLOWED, REDUCE_ONLY, or CASH_PRIORITY
5. **Confidence Level** -- HIGH, MEDIUM, or LOW based on input completeness
### Step 4: Apply Exposure Guidance
Map the posture recommendation to portfolio actions:
| Recommendation | Action |
|----------------|--------|
| NEW_ENTRY_ALLOWED | Proceed with stock-level analysis and new positions |
| REDUCE_ONLY | No new entries; trim existing positions on strength |
| CASH_PRIORITY | Raise cash aggressively; avoid all new commitments |
## Output Format
### JSON Report
```json
{
"schema_version": "1.0",
"generated_at": "2026-03-16T07:00:00Z",
"exposure_ceiling_pct": 70,
"bias": "GROWTH",
"participation": "BROAD",
"recommendation": "NEW_ENTRY_ALLOWED",
"confidence": "HIGH",
"component_scores": {
"breadth_score": 65,
"uptrend_score": 72,
"regime_score": 80,
"top_risk_score": 25,
"ftd_score": 10,
"theme_score": 68,
"sector_score": 70,
"institutional_score": 75
},
"inputs_provided": ["breadth", "uptrend", "regime", "top_risk"],
"inputs_missing": ["ftd", "theme", "sector", "institutional"],
"rationale": "Broad participation with low top risk supports elevated exposure."
}
```
### Markdown Report
The markdown report provides a one-page summary suitable for quick review:
```markdown
# Market Posture Summary
**Date:** 2026-03-16 | **Confidence:** HIGH
## Exposure Ceiling: 70%
| Dimension | Score | Status |
|-----------|-------|--------|
| Breadth | 65 | Healthy |
| Uptrend Participation | 72% | Broad |
| Regime | Broadening | Favorable |
| Top Risk | 25 | Low |
## Recommendation: NEW_ENTRY_ALLOWED
**Bias:** Growth > Value
**Participation:** Broad (healthy internals)
### Rationale
Broad participation with low distribution day count supports elevated equity exposure.
New positions allowed within the 70% ceiling.
```
Reports are saved to `reports/` with filenames `exposure_posture_YYYY-MM-DD_HHMMSS.{json,md}`.
## Resources
- `scripts/calculate_exposure.py` -- Main orchestrator that scores and synthesizes inputs
- `references/exposure_framework.md` -- Scoring rules and threshold definitions
- `references/regime_exposure_map.md` -- Regime-to-exposure ceiling mappings
## Key Principles
1. **Safety First** -- Default to lower exposure when inputs are incomplete or conflicting
2. **Regime Alignment** -- Let macro regime set the baseline; breadth adjusts within bounds
3. **Actionable Output** -- Always produce a clear recommendation, not just data aggregation
## Strict provenance for market-regime-daily
Add `--require-provenance --run-id daily-20261017-unique
--as-of 2026-10-17T13:00:00Z --max-age-days 3` to validate breadth and uptrend
before scoring. Both producers must receive the same unique run ID; pass their
exact printed JSON paths. `--as-of` is the timezone-aware evaluation instant,
not an override of the source business date.
The gate checks schema 1.0, expected run ID, UTC retrieval no later than evaluation,
nonfuture source date and configured age. Default three calendar days is a
conservative policy, not an exchange holiday calendar. Failure stops before
report output. Successful output records accepted lineage and verification limits.
Optional top-risk/other inputs and auxiliary summary dates remain outside this
gate; unknown quality flags stay false. Legacy mode retains its existing schema.
Run IDs are consistency markers, not authenticity or point-in-time proof.