market-breadth-analyzer

어떤 스킬인가요?

Quantifies market breadth health using TraderMonty's public CSV data. Generates a 0-100 composite score across 6 components (100 = healthy). No API key required. Use when user asks about market breadth, participation rate, advance-decline health, whether the rally is broad-based, or general market health assessment.

설치 방법

프로젝트 폴더에서 아래 명령어를 실행하고, 설치할 에이전트를 선택하세요.

설치 명령을 직접 실행해야 적용됩니다. 지원 에이전트와 필요한 권한·라이선스는 제작자의 안내를 확인하세요.

스킬 원문 파일

# Market Breadth Analysis Methodology

## Overview

Market breadth measures the degree of participation in a market move. A healthy advance is characterized by broad participation (many stocks rising), while a narrowing market (fewer stocks leading) often precedes corrections.

This methodology uses TraderMonty's breadth dataset to quantify market health across 6 dimensions, producing a composite score from 0 (critical) to 100 (maximum health).

---

## Data Description

### Breadth Index
The breadth index (0-1) measures the proportion of S&P 500 stocks trading above their 200-day moving average. Key derivatives:
- **Raw:** Daily breadth value (percentage of stocks above 200DMA)
- **8-day EMA (8MA):** Short-term exponentially weighted moving average (fast signal)
- **200-day EMA (200MA):** Long-term exponentially weighted moving average (trend filter)

**Important:** Both moving averages use **EMA** (Exponential Moving Average), not SMA. The source repository calculates them as `ewm(span=N, adjust=False)`. EMA gives more weight to recent data points, making it more responsive to changes than SMA.

### Key Thresholds
| Level | Interpretation |
|-------|---------------|
| 8MA > 0.70 | Very strong breadth - broad rally |
| 8MA > 0.60 | Healthy breadth - above average participation |
| 8MA > 0.50 | Neutral - about half of stocks participating |
| 8MA > 0.40 | Weakening - below average participation |
| 8MA < 0.40 | Extreme weakness - potential trough formation |
| 8MA < 0.20 | Crisis levels - rare, precedes major bottoms |

### Signal Flags
- **Breadth_200MA_Trend:** 1 = 200MA rising (bullish regime), -1 = 200MA falling (bearish regime). Uses hysteresis with threshold=0.001 to prevent whipsaw signals.
- **Bearish_Signal:** Backtested signal combining trend and momentum deterioration
- **Is_Peak / Is_Trough:** Cyclical turning points detected using `scipy.signal.find_peaks` with `distance=50, prominence=0.015` on the 200MA. This ensures at least 50 trading days (~10 weeks) between consecutive peaks/troughs and a minimum prominence of 0.015.
- **Is_Trough_8MA_Below_04:** Extreme trough marker detected with `prominence=0.02` on inverted 8MA values below 0.4.

### Pink Zone (Bearish Region)
The source repository's charts use a pink background to highlight the most dangerous market condition:

```
Pink Zone = (Breadth_200MA_Trend == -1) AND (8MA < 200MA)
```

This means the long-term trend is declining AND the short-term breadth has fallen below the long-term average. Historically, markets in the Pink Zone experience elevated volatility and downside risk. The Pink Zone is distinct from the Bearish_Signal flag - it is a structural condition that can persist for weeks or months.

---

## Component Details

### C1: Current Breadth Level & Trend (25%)

**Rationale:** The most direct measure of current market health. Higher 8MA means more stocks participating; uptrend in 200MA means the long-term structure is supportive.

**Weighting within component:**
- 8MA Level: 70% - immediate health snapshot
- 200MA Trend: 30% - longer-term regime confirmation

**8MA Direction Modifier:** A 5-day lookback on the 8MA itself adjusts the score based on the short-term direction of breadth. This ensures C1 considers not just the level, but whether breadth is accelerating or decelerating:

| Condition | Modifier | Rationale |
|-----------|----------|-----------|
| 8MA falling & level > 0.60 | -10 | Deceleration from a high level — early warning of potential peak |
| 8MA falling & level < 0.40 | +5 | Near-bottom with limited further downside — less penalty |
| 8MA rising & level < 0.60 | +5 | Early recovery bonus — breadth improving from a weak base |
| Otherwise | 0 | No adjustment needed |

**Key insight:** An 8MA of 0.65 in an uptrend (score ~80) is healthier than 0.65 in a downtrend (score ~62), because the downtrend context suggests the level may be transient. The direction modifier further differentiates: a falling 8MA at 0.65 signals emerging weakness (-10), while a rising 8MA at the same level confirms strength.

### C2: 8MA vs 200MA Crossover Dynamics (20%)

**Rationale:** The gap between fast and slow MAs reveals momentum. A wide positive gap means breadth is accelerating above trend; a negative gap means it's deteriorating below trend.

**Direction modifier:** When the 8MA is recovering (rising) while still below the 200MA, this early recovery signal adds +10. Conversely, when 8MA is falling while still above 200MA, this early deterioration signal subtracts -10.

**Key insight:** The crossover point (8MA crossing 200MA) is a significant signal. Bull markets maintain 8MA above 200MA; bear phases see 8MA below 200MA.

### C3: Peak/Trough Cycle Position (20%)

**Rationale:** Breadth moves in cycles. Knowing whether we are in the early, middle, or late phase of a cycle helps calibrate expectations.

**Cycle phases:**
1. **Trough → Early Recovery (0-20 days):** Highest potential for upside if 8MA is rising
2. **Sustained Recovery (21-60 days):** Confirmed recovery with decreasing upside magnitude
3. **Mature Recovery (60+ days):** Late-cycle, watch for next peak
4. **Post-Peak Decline (0-20 days):** Highest risk period
5. **Sustained Decline (21-60 days):** Deep correction territory
6. **Prolonged Decline (60+ days):** Potential bottom formation if 8MA starts rising

**Extreme trough bonus:** When 8MA drops below 0.4 at a trough, history shows these are often excellent long-term entry points, warranting a +10 bonus.

### C4: Bearish Signal Status (15%)

**Rationale:** The dataset includes a backtested bearish signal flag that combines multiple factors. Its value depends on context.

**Interpretation matrix:**
| Signal | Trend | Score | Meaning |
|--------|-------|-------|---------|
| Off | Up | 85 | All clear - no concerns |
| Off | Down | 50 | No immediate danger but bearish backdrop |
| On | Up | 30 | Warning in otherwise bullish environment |
| On | Down | 10 | Full bearish alignment |

**Context matters:** A bearish signal when 8MA is above 0.50 is less concerning (+15 adjustment) than one when 8MA is below 0.25 (-5 adjustment).

**Pink Zone integration:** When in the Pink Zone (200MA downtrend + 8MA below 200MA) but without an active bearish signal, a -10 penalty is applied to reflect structural weakness that the bearish flag alone may not capture.

### C5: Historical Percentile (10%)

**Rationale:** Knowing where current breadth stands relative to the full 10-year history provides framing. Is this level normal, unusually high, or unusually low?

**Overheated/oversold adjustments:** When current 8MA approaches the average peak level, a -10 penalty reflects elevated risk of mean reversion. When near the average trough level, a +10 bonus reflects contrarian opportunity.

### C6: S&P 500 vs Breadth Divergence (10%)

**Rationale:** The most dangerous market condition is when the index makes new highs but breadth is declining (fewer stocks participating). This divergence preceded major tops in 2000, 2007, and 2021.

**Multi-Window Analysis (20d + 60d):** Divergence is measured at two time scales and combined:
- **60-day window (weight 60%):** Captures structural, medium-term divergence
- **20-day window (weight 40%):** Detects short-term emerging divergence

Composite score = 60d_score x 0.6 + 20d_score x 0.4

**Early Warning flag:** When the 20-day window shows bearish divergence (score <= 25) while the 60-day window is still healthy (score >= 50), an Early Warning is triggered. This signals that short-term breadth deterioration has begun before it becomes visible in the structural window.

**Near-flat classification:** When both S&P 500 change and breadth change are below noise thresholds (|S&P %| < 0.5 and |breadth change| < 0.01), the window is classified as "Near-flat (insufficient movement)" with a neutral score of 50. This prevents misclassification of trivial movements as divergence or alignment signals.

**Key patterns (per window):**
- **S&P up + Breadth up:** Healthy, sustainable rally (70)
- **S&P up + Breadth down:** Dangerous narrow market (10-25)
- **S&P down + Breadth up:** Bullish divergence, potential bottom (65-80)
- **S&P down + Breadth down:** Consistent decline, wait for stabilization (30)
- **Both near-flat:** Neutral (50) - insufficient movement to classify

### Weight Redistribution

When a component has `data_available: False` (e.g., insufficient rows for divergence analysis), its weight is excluded and the remaining weights are proportionally redistributed:

```
effective_weight[i] = base_weight[i] / sum(base_weight[j] for all available j)
```

**Example:** If C6 (10%) is unavailable, the remaining 90% is redistributed:
- C1: 25/90 = 27.8%, C2: 20/90 = 22.2%, C3: 20/90 = 22.2%, C4: 15/90 = 16.7%, C5: 10/90 = 11.1%

**Zero-division guard:** If all components are unavailable, the composite score defaults to 50 with zone "Neutral" and a warning that the score is a reference value only.

**Data Quality Labels:**

| Available Components | Label | Interpretation |
|---------------------|-------|----------------|
| 6/6 | Complete | Full confidence |
| 4-5/6 | Partial | Interpret with caution |
| 0-3/6 | Limited | Low confidence |

---

## Composite Score Interpretation

### Zone Thresholds

| Score | Zone | Exposure | Key Actions |
|-------|------|----------|-------------|
| 80-100 | **Strong** | 90-100% | Full position sizing; growth/momentum strategies; wide stops |
| 60-79 | **Healthy** | 75-90% | Normal operations; standard risk management |
| 40-59 | **Neutral** | 60-75% | Selective; tighter stops; avoid speculative names |
| 20-39 | **Weakening** | 40-60% | Profit-taking; raise cash; defensive rotation |
| 0-19 | **Critical** | 25-40% | Capital preservation; hedging; watch for trough |

### Guidance Caution Warning

When the 8MA direction modifier is negative in C1 (Breadth Level & Trend) or C2 (MA Crossover), the report automatically appends a Caution warning to the Overall Assessment guidance. This alerts analysts that the zone-based guidance (e.g., "Normal operations") may be overly optimistic given the deteriorating short-term momentum.

The Caution warning also triggers additional protective actions in the Recommended Actions section:
- "Reduce new position sizes until 8MA stabilizes"
- "Tighten stop-loss levels on existing positions"

These are appended to the standard zone-based actions, not replacing them.

### Cross-Referencing with Other Skills

- **Market Top Detector:** If breadth is Weakening (20-39) AND top detector is Orange/Red, this is strong confirmation of topping conditions.
- **CANSLIM Screener:** In Strong/Healthy zones, CANSLIM stock selections have higher success rates.
- **Sector Analyst:** Weakening breadth often coincides with rotation from offensive to defensive sectors.

### Score History & Trend

The analyzer maintains a rolling history of composite scores to detect trend direction.

**Storage format:** JSON file (`market_breadth_history.json`) in the output directory, structured as an array of objects:

```json
[
  {"data_date": "2025-06-10", "composite_score": 72.3, "component_scores": {"breadth_level_trend": 80, "ma_crossover": 70, "...": "..."}, "recorded_at": "2025-06-10 14:30:00"},
  {"data_date": "2025-06-11", "composite_score": 68.1, "component_scores": {"breadth_level_trend": 75, "ma_crossover": 65, "...": "..."}, "recorded_at": "2025-06-11 09:15:00"}
]
```

**Key rules:**
- **Keyed by data date:** Each entry uses the latest data row's date, not the analysis run date.
- **Duplicate prevention:** If an entry with the same date already exists, it is overwritten (no duplicates).
- **Rolling window:** Maximum 20 entries retained. When the limit is reached, the oldest entry is pruned first.
- **Trend detection:** The delta between the first and last score in the most recent N entries (default N=5) determines the trend label:
  - `delta > 2` → "Improving"
  - `delta < -2` → "Deteriorating"
  - `-2 <= delta <= 2` → "Stable"

**Usage in reports:** The trend summary (`direction`, `delta`, `entries`) is included in both JSON and Markdown outputs, giving analysts a quick read on whether market breadth is accelerating or decelerating.

---

## Historical Context

### Average Values (from Summary CSV)
- **Average Peak (200MA):** ~0.729 - breadth cycles typically top around this level
- **Average Trough (8MA < 0.4):** ~0.232 - extreme troughs average around this level
- **Peak Count:** ~5 over 10 years (roughly every 2 years)
- **Trough Count:** ~10 extreme troughs (roughly twice per year on average)

### Notable Historical Patterns
- **COVID-19 (March 2020):** 8MA crashed to extreme lows, followed by one of the sharpest breadth recoveries in history
- **2022 Bear Market:** Sustained period of 8MA below 200MA with multiple bearish signals
- **2023 Recovery:** Gradual breadth improvement with 8MA crossing above 200MA

---

## Live Resources

- **Interactive Dashboard:** https://tradermonty.github.io/market-breadth-analysis/
- **Data CSV:** https://tradermonty.github.io/market-breadth-analysis/market_breadth_data.csv
- **Summary CSV:** https://tradermonty.github.io/market-breadth-analysis/market_breadth_summary.csv
- **Source Repository:** https://github.com/tradermonty/market-breadth-analysis

Data is automatically updated twice daily via GitHub Actions. CSV files are freely accessible without API keys.

---

## Limitations

1. **Lagging indicator:** Breadth data reflects what has happened, not what will happen. Use alongside forward-looking indicators.
2. **No sector granularity:** The breadth index is market-wide. A few large sectors can dominate the reading.
3. **CSV update frequency:** Data depends on TraderMonty's update schedule. Check data freshness before analysis.
4. **Single market:** Covers S&P 500 only. Does not reflect international markets, small caps, or other asset classes.
5. **No volume context:** The breadth index is price-based and does not incorporate volume data.

# Runtime dependencies for standalone .skill installs (issue #330).
# Verified by scripts/check_skill_deps.py.
requests>=2.31.0

# Market Breadth Analyzer - Calculator modules

#!/usr/bin/env python3
"""
Component 4: Bearish Signal Status (Weight: 15%)

Evaluates the backtested bearish signal flag and its context.
Also detects the "Pink Zone" (Bearish Region) from the source repository:
  Pink Zone = Breadth_200MA_Trend == -1 AND 8MA < 200MA

Input: Bearish_Signal, Breadth_200MA_Trend, Breadth_Index_8MA,
       Breadth_Index_200MA (latest row)

Scoring (100 = healthy):
  Base:
    Signal=False & Trend=1  -> 85 (all clear)
    Signal=False & Trend=-1 -> 50 (no warning but downtrend)
    Signal=True  & Trend=1  -> 30 (warning in uptrend)
    Signal=True  & Trend=-1 -> 10 (fully bearish)

  Context Adjustment:
    Signal=True & 8MA > 0.50 -> +15 (weak signal amid strong breadth)
    Signal=True & 8MA < 0.25 -> -5  (severe signal amid extreme weakness)

  Pink Zone Adjustment:
    In Pink Zone & Signal=False -> -10 (structural weakness even without signal)
"""


def calculate_bearish_signal(rows: list[dict]) -> dict:
    """
    Calculate bearish signal status score.

    Args:
        rows: All detail rows sorted by date ascending.

    Returns:
        Dict with score, signal, and component details.
    """
    if not rows:
        return {
            "score": 50,
            "signal": "NO DATA: No data available for bearish signal analysis",
            "data_available": False,
        }

    latest = rows[-1]
    signal_active = latest["Bearish_Signal"]
    trend = latest["Breadth_200MA_Trend"]
    ma8 = latest["Breadth_Index_8MA"]
    ma200 = latest["Breadth_Index_200MA"]

    # Pink Zone detection (from source repo: chart's pink background)
    # Condition: 200MA trend declining AND 8MA below 200MA
    in_pink_zone = (trend == -1) and (ma8 < ma200)

    # Count consecutive pink zone days
    pink_zone_days = 0
    if in_pink_zone:
        for i in range(len(rows) - 1, -1, -1):
            r = rows[i]
            if r["Breadth_200MA_Trend"] == -1 and r["Breadth_Index_8MA"] < r["Breadth_Index_200MA"]:
                pink_zone_days += 1
            else:
                break

    # Base score
    base_score = _base_score(signal_active, trend)

    # Context adjustment
    context_adj = 0
    if signal_active:
        if ma8 > 0.50:
            context_adj = +15  # Weak signal amid strong breadth
        elif ma8 < 0.25:
            context_adj = -5  # Severe signal amid extreme weakness

    # Pink Zone adjustment: structural weakness even without bearish signal
    pink_zone_adj = 0
    if in_pink_zone and not signal_active:
        pink_zone_adj = -10

    score = round(base_score + context_adj + pink_zone_adj)
    score = max(0, min(100, score))

    signal_text = _generate_signal(signal_active, trend, ma8, in_pink_zone, score)

    return {
        "score": score,
        "signal": signal_text,
        "data_available": True,
        "signal_active": signal_active,
        "trend": trend,
        "current_8ma": ma8,
        "current_200ma": ma200,
        "in_pink_zone": in_pink_zone,
        "pink_zone_days": pink_zone_days,
        "base_score": base_score,
        "context_adjustment": context_adj,
        "pink_zone_adjustment": pink_zone_adj,
        "date": latest["Date"],
    }


def _base_score(signal_active: bool, trend: int) -> int:
    """Calculate base score from signal and trend combination."""
    if not signal_active and trend == 1:
        return 85  # All clear
    elif not signal_active and trend == -1:
        return 50  # No warning but downtrend
    elif signal_active and trend == 1:
        return 30  # Warning in uptrend
    else:  # signal_active and trend == -1
        return 10  # Fully bearish


def _generate_signal(
    signal_active: bool,
    trend: int,
    ma8: float,
    in_pink_zone: bool,
    score: int,
) -> str:
    """Generate human-readable signal."""
    trend_str = "uptrend" if trend == 1 else "downtrend"
    pink_str = " [PINK ZONE]" if in_pink_zone else ""

    if not signal_active:
        if trend == 1:
            return f"ALL CLEAR: No bearish signal, {trend_str}"
        elif in_pink_zone:
            return (
                f"CAUTION: No bearish signal but in Pink Zone (downtrend + 8MA < 200MA){pink_str}"
            )
        else:
            return f"CAUTION: No bearish signal but {trend_str}"
    else:
        if ma8 > 0.50:
            return (
                f"WARNING (muted): Bearish signal active in {trend_str}, "
                f"but 8MA={ma8:.3f} still relatively strong{pink_str}"
            )
        elif ma8 < 0.25:
            return (
                f"CRITICAL: Bearish signal active in {trend_str}, "
                f"8MA={ma8:.3f} extremely weak{pink_str}"
            )
        else:
            return f"BEARISH: Signal active in {trend_str}, 8MA={ma8:.3f}{pink_str}"

#!/usr/bin/env python3
"""
Component 3: Peak/Trough Cycle Position (Weight: 20%)

Determines where we are in the breadth cycle relative to recent peaks and troughs.

Input: Is_Peak, Is_Trough, Is_Trough_8MA_Below_04, Breadth_Index_8MA (last 120 days)

Scoring (100 = healthy):
  Latest marker = TROUGH:
    <= 20 days & 8MA rising       -> 85 (early recovery)
    <= 20 days & 8MA flat/falling -> 30 (failed reversal)
    21-60 days & 8MA rising       -> 75 (sustained recovery)
    > 60 days & 8MA rising        -> 65 (mature recovery)

  Latest marker = PEAK:
    <= 20 days & 8MA falling      -> 20 (post-peak decline)
    <= 20 days & 8MA flat/rising  -> 60 (high-level consolidation)
    21-60 days & 8MA falling      -> 15 (sustained decline)
    > 60 days & 8MA falling       -> 10 (prolonged decline)
    > 60 days & 8MA rising        -> 50 (possible bottom formation)

  Extreme Trough Bonus: Is_Trough_8MA_Below_04 == True -> +10
  No marker within 120 days -> 50 (neutral)
"""

from typing import Optional


def calculate_cycle_position(rows: list[dict]) -> dict:
    """
    Calculate peak/trough cycle position score.

    Args:
        rows: All detail rows sorted by date ascending.

    Returns:
        Dict with score, signal, and component details.
    """
    if not rows or len(rows) < 10:
        return {
            "score": 50,
            "signal": "NO DATA: Insufficient data for cycle analysis",
            "data_available": False,
        }

    # Look at last 120 trading days
    lookback = min(120, len(rows))
    recent = rows[-lookback:]
    latest = rows[-1]
    current_8ma = latest["Breadth_Index_8MA"]

    # Find latest peak or trough marker
    marker_type, marker_idx, days_since = _find_latest_marker(recent)

    # Determine 8MA trend (rising/falling) over last 5 days
    if len(rows) >= 6:
        ma8_trend = "rising" if current_8ma > rows[-6]["Breadth_Index_8MA"] else "falling"
    else:
        ma8_trend = "unknown"

    # Check for extreme trough
    extreme_trough = False
    if marker_type == "TROUGH":
        extreme_trough = recent[marker_idx]["Is_Trough_8MA_Below_04"]

    # Calculate score
    score = _calculate_score(marker_type, days_since, ma8_trend, extreme_trough)
    score = max(0, min(100, score))

    signal = _generate_signal(marker_type, days_since, ma8_trend, score)

    marker_found = marker_type is not None

    return {
        "score": score,
        "signal": signal,
        "data_available": marker_found,
        "latest_marker_type": marker_type,
        "days_since_marker": days_since,
        "ma8_trend": ma8_trend,
        "current_8ma": current_8ma,
        "extreme_trough": extreme_trough,
        "date": latest["Date"],
    }


def _find_latest_marker(recent: list[dict]) -> tuple[Optional[str], int, Optional[int]]:
    """
    Find the most recent peak or trough marker.

    Returns (marker_type, index_in_recent, days_since).
    """
    for i in range(len(recent) - 1, -1, -1):
        row = recent[i]
        if row["Is_Peak"]:
            days_since = len(recent) - 1 - i
            return "PEAK", i, days_since
        if row["Is_Trough"] or row["Is_Trough_8MA_Below_04"]:
            days_since = len(recent) - 1 - i
            return "TROUGH", i, days_since

    return None, -1, None


def _calculate_score(
    marker_type: Optional[str],
    days_since: Optional[int],
    ma8_trend: str,
    extreme_trough: bool,
) -> int:
    """Calculate cycle position score based on marker and context."""
    if marker_type is None or days_since is None:
        return 50  # Neutral when no marker found

    rising = ma8_trend == "rising"

    if marker_type == "TROUGH":
        if days_since <= 20:
            base = 85 if rising else 30
        elif days_since <= 60:
            base = 75 if rising else 35
        else:
            base = 65 if rising else 40

        # Extreme trough bonus (contrarian buy signal)
        if extreme_trough:
            base += 10

    elif marker_type == "PEAK":
        if days_since <= 20:
            base = 60 if rising else 20
        elif days_since <= 60:
            base = 45 if rising else 15
        else:
            base = 50 if rising else 10

    else:
        base = 50

    return base


def _generate_signal(
    marker_type: Optional[str],
    days_since: Optional[int],
    ma8_trend: str,
    score: int,
) -> str:
    """Generate human-readable signal."""
    if marker_type is None:
        return "NEUTRAL: No cycle marker in last 120 days"

    if marker_type == "TROUGH":
        if days_since <= 20:
            phase = "early recovery" if ma8_trend == "rising" else "failed reversal attempt"
        elif days_since <= 60:
            phase = "sustained recovery" if ma8_trend == "rising" else "stalled recovery"
        else:
            phase = "mature recovery" if ma8_trend == "rising" else "weakening recovery"
        return f"TROUGH ({days_since}d ago): {phase}, 8MA {ma8_trend}"

    else:  # PEAK
        if days_since <= 20:
            phase = "consolidation near highs" if ma8_trend == "rising" else "post-peak decline"
        elif days_since <= 60:
            phase = "gradual decline" if ma8_trend != "rising" else "recovery attempt"
        else:
            phase = "possible bottom formation" if ma8_trend == "rising" else "prolonged decline"
        return f"PEAK ({days_since}d ago): {phase}, 8MA {ma8_trend}"

#!/usr/bin/env python3
"""
Component 6: S&P 500 vs Breadth Divergence (Weight: 10%)

Detects divergence between price action and breadth participation.

Uses dual windows:
  - 60-day window (weight 0.6): Medium-term structural divergence
  - 20-day window (weight 0.4): Short-term early warning

Composite score = 60d_score * 0.6 + 20d_score * 0.4

Early Warning: 20d bearish (<=25) while 60d healthy (>=50) flags
an emerging short-term divergence before it becomes structural.

Input: S&P500_Price, Breadth_Index_8MA (last 60+ days)

Scoring per window (100 = healthy):
  Both rising                        -> 70 (healthy rally)
  Both falling                       -> 30 (consistent decline)
  SP up(>3%) & Breadth down(<-0.05)  -> 10 (dangerous divergence)
  SP up(>1%) & Breadth down(<-0.03)  -> 25
  SP down(<-3%) & Breadth up(>+0.05) -> 80 (bullish divergence)
  SP down(<-1%) & Breadth up(>+0.03) -> 65
  Otherwise                          -> 50
"""


def calculate_divergence(rows: list[dict]) -> dict:
    """
    Calculate S&P 500 vs breadth divergence score using dual windows.

    Args:
        rows: All detail rows sorted by date ascending.

    Returns:
        Dict with score, signal, windows, and component details.
    """
    if not rows or len(rows) < 20:
        return {
            "score": 50,
            "signal": "NO DATA: Insufficient data for divergence analysis",
            "data_available": False,
        }

    latest = rows[-1]

    # Compute both windows
    w60 = _compute_window(rows, 60)
    w20 = _compute_window(rows, 20)

    # Composite score
    score = round(w60["score"] * 0.6 + w20["score"] * 0.4, 1)
    score = max(0, min(100, score))

    # Early Warning: short-term bearish divergence while long-term healthy
    early_warning = w20["score"] <= 25 and w60["score"] >= 50

    # Signal uses the composite
    signal = _generate_signal(w60["sp_pct"], w60["breadth_chg"], w60["div_type"], score)

    return {
        "score": score,
        "signal": signal,
        "data_available": True,
        # Top-level backward compatibility (from 60d window)
        "sp500_pct_change": round(w60["sp_pct"], 2),
        "breadth_change": round(w60["breadth_chg"], 4),
        "sp500_latest": latest["S&P500_Price"],
        "sp500_past": w60["sp_past"],
        "ma8_latest": latest["Breadth_Index_8MA"],
        "ma8_past": w60["ma8_past"],
        "lookback_days": w60["lookback_days"],
        "divergence_type": w60["div_type"],
        "date": latest["Date"],
        # New window details
        "window_60d": {
            "score": w60["score"],
            "divergence_type": w60["div_type"],
            "sp500_pct_change": round(w60["sp_pct"], 2),
            "breadth_change": round(w60["breadth_chg"], 4),
            "lookback_days": w60["lookback_days"],
        },
        "window_20d": {
            "score": w20["score"],
            "divergence_type": w20["div_type"],
            "sp500_pct_change": round(w20["sp_pct"], 2),
            "breadth_change": round(w20["breadth_chg"], 4),
            "lookback_days": w20["lookback_days"],
        },
        "early_warning": early_warning,
    }


def _compute_window(rows: list[dict], lookback: int) -> dict:
    """Compute divergence metrics for a single lookback window."""
    actual_lookback = min(lookback, len(rows))
    latest = rows[-1]
    past = rows[-actual_lookback]

    sp_latest = latest["S&P500_Price"]
    sp_past = past["S&P500_Price"]
    ma8_latest = latest["Breadth_Index_8MA"]
    ma8_past = past["Breadth_Index_8MA"]

    if sp_past <= 0:
        return {
            "score": 50,
            "sp_pct": 0.0,
            "breadth_chg": 0.0,
            "div_type": "Invalid data",
            "sp_past": sp_past,
            "ma8_past": ma8_past,
            "lookback_days": actual_lookback,
        }

    sp_pct = (sp_latest - sp_past) / sp_past * 100
    breadth_chg = ma8_latest - ma8_past

    score, div_type = _score_divergence(sp_pct, breadth_chg)
    score = max(0, min(100, score))

    return {
        "score": score,
        "sp_pct": sp_pct,
        "breadth_chg": breadth_chg,
        "div_type": div_type,
        "sp_past": sp_past,
        "ma8_past": ma8_past,
        "lookback_days": actual_lookback,
    }


def _score_divergence(sp_pct: float, breadth_chg: float) -> tuple[int, str]:
    """Score based on price/breadth divergence. Returns (score, type_label)."""
    sp_up = sp_pct > 0
    breadth_up = breadth_chg > 0

    # Dangerous divergence: SP up, breadth down
    if sp_pct > 3.0 and breadth_chg < -0.05:
        return 10, "Dangerous bearish divergence"
    if sp_pct > 1.0 and breadth_chg < -0.03:
        return 25, "Moderate bearish divergence"

    # Bullish divergence: SP down, breadth up
    if sp_pct < -3.0 and breadth_chg > 0.05:
        return 80, "Strong bullish divergence"
    if sp_pct < -1.0 and breadth_chg > 0.03:
        return 65, "Moderate bullish divergence"

    # Near-flat movements (noise level)
    if abs(sp_pct) < 0.5 and abs(breadth_chg) < 0.01:
        return 50, "Near-flat (insufficient movement)"

    # Aligned movements
    if sp_up and breadth_up:
        return 70, "Healthy alignment (both rising)"
    if not sp_up and not breadth_up:
        return 30, "Consistent decline (both falling)"

    return 50, "Mixed signals"


def _generate_signal(sp_pct: float, breadth_chg: float, div_type: str, score: int) -> str:
    """Generate human-readable signal."""
    return f"{div_type}: S&P {sp_pct:+.1f}%, Breadth 8MA {breadth_chg:+.3f} over 60d"

#!/usr/bin/env python3
"""
Component 5: Historical Percentile (Weight: 10%)

Determines where the current breadth level sits in the full historical distribution.

Input: Breadth_Index_8MA (all rows), Summary CSV (average peak/trough values)

Scoring (100 = healthy):
  Percentile rank of current 8MA among all historical 8MA values:
    >= 80th -> 90    >= 60th -> 70    >= 40th -> 50
    >= 20th -> 30    <  20th -> 10

  Adjustment:
    Current >= avg_peak * 0.95 -> -10 (overheated caution)
    Current <= avg_trough * 1.05 -> +10 (oversold zone)
"""

from typing import Optional


def calculate_historical_percentile(
    rows: list[dict],
    summary: dict[str, str],
) -> dict:
    """
    Calculate historical percentile score.

    Args:
        rows: All detail rows sorted by date ascending.
        summary: Summary CSV as dict {Metric: Value}.

    Returns:
        Dict with score, signal, and component details.
    """
    if not rows:
        return {
            "score": 50,
            "signal": "NO DATA: No data available for historical analysis",
            "data_available": False,
        }

    # Extract all 8MA values
    all_8ma = [r["Breadth_Index_8MA"] for r in rows]
    current_8ma = all_8ma[-1]

    # Calculate percentile rank
    below_count = sum(1 for v in all_8ma if v < current_8ma)
    percentile = (below_count / len(all_8ma)) * 100

    # Base score from percentile
    base_score = _score_percentile(percentile)

    # Parse summary values
    avg_peak = _safe_float(summary.get("Average Peaks (200MA)", ""))
    avg_trough = _safe_float(summary.get("Average Troughs (8MA < 0.4)", ""))

    # Adjustment for extremes
    adjustment = 0
    if avg_peak is not None and current_8ma >= avg_peak * 0.95:
        adjustment = -10  # Overheated
    elif avg_trough is not None and current_8ma <= avg_trough * 1.05:
        adjustment = +10  # Oversold

    score = round(base_score + adjustment)
    score = max(0, min(100, score))

    signal = _generate_signal(percentile, adjustment, score)

    return {
        "score": score,
        "signal": signal,
        "data_available": True,
        "current_8ma": current_8ma,
        "percentile_rank": percentile,
        "base_score": base_score,
        "adjustment": adjustment,
        "avg_peak": avg_peak,
        "avg_trough": avg_trough,
        "total_observations": len(all_8ma),
        "date": rows[-1]["Date"],
    }


def _score_percentile(percentile: float) -> int:
    """Score from percentile rank."""
    if percentile >= 80:
        return 90
    elif percentile >= 60:
        return 70
    elif percentile >= 40:
        return 50
    elif percentile >= 20:
        return 30
    else:
        return 10


def _safe_float(val: str) -> Optional[float]:
    """Safely parse float from string."""
    try:
        return float(val)
    except (ValueError, TypeError):
        return None


def _generate_signal(percentile: float, adjustment: int, score: int) -> str:
    """Generate human-readable signal."""
    adj_str = ""
    if adjustment < 0:
        adj_str = " (near historical peak - overheated)"
    elif adjustment > 0:
        adj_str = " (near historical trough - oversold)"

    if score >= 70:
        return f"HIGH: {percentile:.0f}th percentile - above average breadth{adj_str}"
    elif score >= 40:
        return f"AVERAGE: {percentile:.0f}th percentile - normal range{adj_str}"
    else:
        return f"LOW: {percentile:.0f}th percentile - below average breadth{adj_str}"

#!/usr/bin/env python3
"""
Component 2: 8MA vs 200MA Crossover Dynamics (Weight: 20%)

Detects momentum and direction changes via the gap between fast and slow MAs.

Input: Breadth_Index_8MA, Breadth_Index_200MA (latest + 5-day lookback)

Scoring (100 = healthy):
  Gap = 8MA - 200MA
    >= +0.15 -> 95    >= +0.10 -> 80    >= +0.05 -> 65
    >=  0.00 -> 50    >= -0.05 -> 35    >= -0.10 -> 20
    <  -0.10 -> 5

  Direction Modifier (5-day 8MA trend):
    Gap < 0 and 8MA rising  -> +10 (recovery signal)
    Gap > 0 and 8MA falling -> -10 (deterioration signal)
"""


def calculate_ma_crossover(rows: list[dict]) -> dict:
    """
    Calculate 8MA vs 200MA crossover dynamics score.

    Args:
        rows: All detail rows sorted by date ascending.

    Returns:
        Dict with score, signal, and component details.
    """
    if not rows or len(rows) < 6:
        return {
            "score": 50,
            "signal": "NO DATA: Insufficient data for crossover analysis",
            "data_available": False,
        }

    latest = rows[-1]
    ma8 = latest["Breadth_Index_8MA"]
    ma200 = latest["Breadth_Index_200MA"]
    gap = ma8 - ma200

    # Gap score
    gap_score = _score_gap(gap)

    # 8MA direction over last 5 days
    ma8_5d_ago = rows[-6]["Breadth_Index_8MA"]
    ma8_rising = ma8 > ma8_5d_ago
    ma8_direction = "rising" if ma8_rising else "falling"

    # Direction modifier
    direction_modifier = 0
    if gap < 0 and ma8_rising:
        direction_modifier = +10  # Recovery signal
    elif gap > 0 and not ma8_rising:
        direction_modifier = -10  # Deterioration signal

    score = round(gap_score + direction_modifier)
    score = max(0, min(100, score))

    signal = _generate_signal(gap, ma8_direction, direction_modifier, score)

    return {
        "score": score,
        "signal": signal,
        "data_available": True,
        "gap": gap,
        "gap_score": gap_score,
        "current_8ma": ma8,
        "current_200ma": ma200,
        "ma8_5d_ago": ma8_5d_ago,
        "ma8_direction": ma8_direction,
        "direction_modifier": direction_modifier,
        "date": latest["Date"],
    }


def _score_gap(gap: float) -> int:
    """Score based on 8MA - 200MA gap."""
    if gap >= 0.15:
        return 95
    elif gap >= 0.10:
        return 80
    elif gap >= 0.05:
        return 65
    elif gap >= 0.00:
        return 50
    elif gap >= -0.05:
        return 35
    elif gap >= -0.10:
        return 20
    else:
        return 5


def _generate_signal(gap: float, direction: str, modifier: int, score: int) -> str:
    """Generate human-readable signal."""
    mod_str = ""
    if modifier > 0:
        mod_str = " (recovery signal)"
    elif modifier < 0:
        mod_str = " (deterioration signal)"

    if score >= 80:
        return f"BULLISH: 8MA well above 200MA (gap={gap:+.3f}, 8MA {direction}){mod_str}"
    elif score >= 60:
        return f"POSITIVE: 8MA above 200MA (gap={gap:+.3f}, 8MA {direction}){mod_str}"
    elif score >= 40:
        return f"NEUTRAL: Near crossover (gap={gap:+.3f}, 8MA {direction}){mod_str}"
    elif score >= 20:
        return f"NEGATIVE: 8MA below 200MA (gap={gap:+.3f}, 8MA {direction}){mod_str}"
    else:
        return f"BEARISH: 8MA far below 200MA (gap={gap:+.3f}, 8MA {direction}){mod_str}"

#!/usr/bin/env python3
"""
Component 1: Current Breadth Level & Trend (Weight: 25%)

The most direct indicator of market health.

Input: Breadth_Index_8MA, Breadth_200MA_Trend (latest row + 5-day lookback)

Scoring (100 = healthy):
  8MA Level (70% of component):
    >= 0.70 -> 95    >= 0.60 -> 80    >= 0.50 -> 65
    >= 0.40 -> 50    >= 0.30 -> 35    >= 0.20 -> 20
    <  0.20 -> 5

  200MA Trend (30% of component):
    Trend == 1  -> 80 (uptrend)
    Trend == -1 -> 20 (downtrend)

  8MA Direction Modifier (5-day direction):
    Falling & level > 0.60 -> -10 (deceleration from high level)
    Falling & level < 0.40 -> +5  (limited downside near bottom)
    Rising  & level < 0.60 -> +5  (early recovery bonus)
    Otherwise              ->  0

  Score = clamp(0.70 * level_score + 0.30 * trend_score + modifier, 0, 100)
"""


def calculate_breadth_level_trend(rows: list[dict]) -> dict:
    """
    Calculate current breadth level and trend score.

    Args:
        rows: All detail rows sorted by date ascending.

    Returns:
        Dict with score, signal, and component details.
    """
    if not rows:
        return {
            "score": 50,
            "signal": "NO DATA: No breadth data available",
            "data_available": False,
        }

    latest = rows[-1]
    ma8 = latest["Breadth_Index_8MA"]
    trend = latest["Breadth_200MA_Trend"]

    # 8MA level score
    level_score = _score_8ma_level(ma8)

    # Trend score
    trend_score = 80 if trend == 1 else 20

    # 8MA direction modifier
    ma8_direction, direction_modifier = _direction_modifier(rows, ma8)

    # Combined
    base = round(0.70 * level_score + 0.30 * trend_score)
    score = max(0, min(100, base + direction_modifier))

    # Signal
    signal = _generate_signal(ma8, trend, score)

    result = {
        "score": score,
        "signal": signal,
        "data_available": True,
        "current_8ma": ma8,
        "current_200ma": latest["Breadth_Index_200MA"],
        "trend": trend,
        "level_score": level_score,
        "trend_score": trend_score,
        "date": latest["Date"],
    }

    if ma8_direction is not None:
        result["ma8_direction"] = ma8_direction
        result["direction_modifier"] = direction_modifier

    return result


def _direction_modifier(rows: list[dict], current_8ma: float):
    """Calculate 8MA direction modifier based on 5-day movement.

    Returns (direction_str_or_None, modifier_int).
    """
    if len(rows) < 6:
        return None, 0

    ma8_5d_ago = rows[-6]["Breadth_Index_8MA"]

    if current_8ma > ma8_5d_ago:
        direction = "rising"
    elif current_8ma < ma8_5d_ago:
        direction = "falling"
    else:
        return "flat", 0

    if direction == "falling" and current_8ma > 0.60:
        return direction, -10
    if direction == "falling" and current_8ma < 0.40:
        return direction, 5
    if direction == "rising" and current_8ma < 0.60:
        return direction, 5

    return direction, 0


def _score_8ma_level(ma8: float) -> int:
    """Score based on 8MA level."""
    if ma8 >= 0.70:
        return 95
    elif ma8 >= 0.60:
        return 80
    elif ma8 >= 0.50:
        return 65
    elif ma8 >= 0.40:
        return 50
    elif ma8 >= 0.30:
        return 35
    elif ma8 >= 0.20:
        return 20
    else:
        return 5


def _generate_signal(ma8: float, trend: int, score: int) -> str:
    """Generate human-readable signal."""
    trend_str = "uptrend" if trend == 1 else "downtrend"

    if score >= 80:
        return f"STRONG: 8MA={ma8:.3f} in {trend_str} - broad participation"
    elif score >= 60:
        return f"HEALTHY: 8MA={ma8:.3f} in {trend_str} - above average"
    elif score >= 40:
        return f"NEUTRAL: 8MA={ma8:.3f} in {trend_str} - mixed signals"
    elif score >= 20:
        return f"WEAK: 8MA={ma8:.3f} in {trend_str} - deteriorating breadth"
    else:
        return f"CRITICAL: 8MA={ma8:.3f} in {trend_str} - severe weakness"

#!/usr/bin/env python3
"""
Market Breadth Analyzer - CSV Data Client

Fetches and parses market breadth CSV data from TraderMonty's public GitHub Pages.
No API key required.

Data sources:
  Detail:  https://tradermonty.github.io/market-breadth-analysis/market_breadth_data.csv
  Summary: https://tradermonty.github.io/market-breadth-analysis/market_breadth_summary.csv
"""

import csv
import io
import math
import sys
from datetime import datetime
from typing import Optional

import requests

DEFAULT_DETAIL_URL = "https://tradermonty.github.io/market-breadth-analysis/market_breadth_data.csv"
DEFAULT_SUMMARY_URL = (
    "https://tradermonty.github.io/market-breadth-analysis/market_breadth_summary.csv"
)

DETAIL_COLUMNS = {
    "Date": str,
    "S&P500_Price": float,
    "Breadth_Index_Raw": float,
    "Breadth_Index_200MA": float,
    "Breadth_Index_8MA": float,
    "Breadth_200MA_Trend": int,
    "Bearish_Signal": bool,
    "Is_Peak": bool,
    "Is_Trough": bool,
    "Is_Trough_8MA_Below_04": bool,
}

TIMEOUT = 30

SUMMARY_COLUMNS = ("Metric", "Value")
REQUIRED_SUMMARY_METRICS = ("Average Peaks (200MA)", "Average Troughs (8MA < 0.4)")


def _open_reader(text: str) -> csv.DictReader:
    """Build a DictReader with a BOM-free text and whitespace-stripped header names.

    DictReader row keys come from ``fieldnames``, so stripping them once here keeps
    the header check and the row lookups consistent.
    """
    reader = csv.DictReader(io.StringIO(text.lstrip("\ufeff")))
    if reader.fieldnames is not None:
        reader.fieldnames = [(name or "").strip() for name in reader.fieldnames]
    return reader


def _read_csv(text: str, filename: str) -> Optional[tuple[list[str], list[dict]]]:
    """Parse a whole CSV body; report an unparsable body (NUL, oversized field) once."""
    try:
        reader = _open_reader(text)
        return list(reader.fieldnames or []), list(reader)
    except csv.Error as exc:
        print("FAILED")
        print(f"ERROR: {filename} schema drift: unparsable CSV ({exc})", file=sys.stderr)
        return None


def _missing_columns(fieldnames, required) -> list[str]:
    """Return sorted required columns absent from a header normalized by _open_reader."""
    return sorted(set(required) - set(fieldnames or []))


def _report_drift(filename: str, missing: list[str]) -> None:
    print(
        f"ERROR: {filename} schema drift: missing columns [{','.join(missing)}]",
        file=sys.stderr,
    )


def fetch_detail_csv(url: str = DEFAULT_DETAIL_URL) -> list[dict]:
    """
    Fetch and parse the detail CSV (market_breadth_data.csv).

    Returns list of dicts sorted by date ascending (oldest first).
    """
    print(f"  Fetching detail CSV from {url}...", end=" ", flush=True)
    try:
        resp = requests.get(url, timeout=TIMEOUT)
        resp.raise_for_status()
    except requests.RequestException as e:
        print("FAILED")
        print(f"ERROR: Failed to fetch detail CSV: {e}", file=sys.stderr)
        return []

    parsed = _read_csv(resp.text, "market_breadth_data.csv")
    if parsed is None:
        return []
    fieldnames, raw_rows = parsed
    missing = _missing_columns(fieldnames, DETAIL_COLUMNS)
    if missing:
        print("FAILED")
        _report_drift("market_breadth_data.csv", missing)
        return []

    rows = []

    for line_num, raw_row in enumerate(raw_rows, start=2):
        try:
            row = _parse_detail_row(raw_row)
            rows.append(row)
        except (ValueError, KeyError) as e:
            print(f"\n  WARN: Skipping line {line_num}: {e}", file=sys.stderr)

    if not rows:
        print("OK (0 rows)")
        return []

    # Sort by date ascending
    rows.sort(key=lambda r: r["Date"])

    print(f"OK ({len(rows)} rows, {rows[0]['Date']} to {rows[-1]['Date']})")
    return rows


def fetch_summary_csv(url: str = DEFAULT_SUMMARY_URL) -> dict[str, str]:
    """
    Fetch and parse the summary CSV (market_breadth_summary.csv).

    Returns dict mapping Metric -> Value.
    """
    print(f"  Fetching summary CSV from {url}...", end=" ", flush=True)
    try:
        resp = requests.get(url, timeout=TIMEOUT)
        resp.raise_for_status()
    except requests.RequestException as e:
        print("FAILED")
        print(f"ERROR: Failed to fetch summary CSV: {e}", file=sys.stderr)
        return {}

    parsed = _read_csv(resp.text, "market_breadth_summary.csv")
    if parsed is None:
        return {}
    fieldnames, raw_rows = parsed
    missing = _missing_columns(fieldnames, SUMMARY_COLUMNS)
    if missing:
        print("FAILED")
        _report_drift("market_breadth_summary.csv", missing)
        return {}

    summary = {}
    for line_num, raw_row in enumerate(raw_rows, start=2):
        raw_metric = raw_row.get("Metric")
        raw_value = raw_row.get("Value")
        if raw_metric is None or raw_value is None:
            print(f"\n  WARN: Skipping summary line {line_num}: truncated row", file=sys.stderr)
            continue
        metric = raw_metric.strip()
        value = raw_value.strip()
        if metric:
            summary[metric] = value

    print(f"OK ({len(summary)} metrics)")
    for name in REQUIRED_SUMMARY_METRICS:
        if name not in summary:
            print(f"WARN: market_breadth_summary.csv missing metric '{name}'", file=sys.stderr)
    return summary


def check_data_freshness(
    rows: list[dict], max_stale_days: int = 5, detail_url: str = DEFAULT_DETAIL_URL
) -> dict:
    """
    Check whether the latest data is reasonably fresh.
    Uses both CSV date and HTTP Last-Modified header for accuracy.

    Args:
        rows: Parsed detail rows (sorted by date ascending).
        max_stale_days: Calendar days before data is considered stale.
        detail_url: URL to check Last-Modified header against.

    Returns:
        Dict with is_fresh, latest_date, days_old, last_modified, warning.
    """
    if not rows:
        return {
            "is_fresh": False,
            "latest_date": None,
            "days_old": None,
            "last_modified": None,
            "warning": "No data available",
        }

    latest_date_str = rows[-1]["Date"]
    try:
        latest_date = datetime.strptime(latest_date_str, "%Y-%m-%d")
    except ValueError:
        return {
            "is_fresh": False,
            "latest_date": latest_date_str,
            "days_old": None,
            "last_modified": None,
            "warning": f"Cannot parse latest date: {latest_date_str}",
        }

    days_old = (datetime.now() - latest_date).days
    is_fresh = days_old <= max_stale_days

    # Check HTTP Last-Modified header for additional freshness info
    last_modified = _check_last_modified(detail_url)

    warning = None
    if last_modified is None:
        warning = "Degraded freshness check: HTTP Last-Modified unavailable; using CSV date only."
    if not is_fresh:
        warning = (
            (warning + " " if warning else "")
            + f"Data is {days_old} days old (latest: {latest_date_str}). "
            f"Threshold: {max_stale_days} days. Results may not reflect current conditions."
        )

    return {
        "is_fresh": is_fresh,
        "latest_date": latest_date_str,
        "days_old": days_old,
        "last_modified": last_modified,
        "warning": warning,
    }


def _check_last_modified(url: str) -> Optional[str]:
    """Check HTTP Last-Modified header via HEAD request."""
    try:
        resp = requests.head(url, timeout=10, allow_redirects=True)
        resp.raise_for_status()
        lm = resp.headers.get("Last-Modified")
        if lm:
            from email.utils import parsedate_to_datetime

            dt = parsedate_to_datetime(lm)
            return dt.strftime("%Y-%m-%d %H:%M UTC")
    except (requests.RequestException, ValueError, TypeError, OverflowError):
        print("WARN: Degraded freshness check; HTTP Last-Modified unavailable", file=sys.stderr)
    return None


def get_latest_n_rows(rows: list[dict], n: int) -> list[dict]:
    """Return the last n rows (most recent)."""
    return rows[-n:] if len(rows) >= n else rows[:]


def _parse_detail_row(raw: dict) -> dict:
    """Convert a raw CSV row dict to properly typed values."""
    missing = [column for column in DETAIL_COLUMNS if raw.get(column) is None]
    if missing:
        raise ValueError(f"missing values for: {', '.join(missing)}")

    row = {}
    row["Date"] = raw["Date"].strip()
    try:
        parsed_date = datetime.strptime(row["Date"], "%Y-%m-%d")
    except ValueError as exc:
        raise ValueError(f"invalid Date: {row['Date']!r}") from exc
    if parsed_date.strftime("%Y-%m-%d") != row["Date"]:
        raise ValueError(f"invalid Date: {row['Date']!r}")
    row["S&P500_Price"] = _parse_finite_float(raw["S&P500_Price"], "S&P500_Price")
    row["Breadth_Index_Raw"] = _parse_finite_float(raw["Breadth_Index_Raw"], "Breadth_Index_Raw")
    row["Breadth_Index_200MA"] = _parse_finite_float(
        raw["Breadth_Index_200MA"], "Breadth_Index_200MA"
    )
    row["Breadth_Index_8MA"] = _parse_finite_float(raw["Breadth_Index_8MA"], "Breadth_Index_8MA")
    row["Breadth_200MA_Trend"] = int(raw["Breadth_200MA_Trend"])
    row["Bearish_Signal"] = _parse_bool(raw["Bearish_Signal"])
    row["Is_Peak"] = _parse_bool(raw["Is_Peak"])
    row["Is_Trough"] = _parse_bool(raw["Is_Trough"])
    row["Is_Trough_8MA_Below_04"] = _parse_bool(raw["Is_Trough_8MA_Below_04"])
    return row


def _parse_finite_float(val: str, column: str) -> float:
    """Parse a CSV number and reject NaN/Infinity before scoring."""
    parsed = float(val)
    if not math.isfinite(parsed):
        raise ValueError(f"{column} must be finite")
    return parsed


def _parse_bool(val: str) -> bool:
    """Parse boolean from CSV string."""
    normalized = val.strip().lower()
    if normalized in ("true", "1", "yes"):
        return True
    if normalized in ("false", "0", "no"):
        return False
    raise ValueError(f"invalid boolean value: {val!r}")


# Testing
if __name__ == "__main__":
    print("Testing CSV Client...\n")

    detail = fetch_detail_csv()
    print(f"  Detail rows: {len(detail)}")
    if detail:
        print(f"  First row: {detail[0]}")
        print(f"  Last row:  {detail[-1]}")

    print()
    summary = fetch_summary_csv()
    print(f"  Summary metrics: {summary}")

    print()
    freshness = check_data_freshness(detail)
    print(f"  Freshness: {freshness}")

    print("\nAll tests completed.")

#!/usr/bin/env python3
"""
Market Breadth Analyzer - Score History Tracker

Persists composite scores keyed by data date (the latest date in the CSV),
enabling trend analysis across consecutive runs.

History file format: JSON array of entries, each with:
  - data_date: str (YYYY-MM-DD) — the date of the analyzed data
  - composite_score: float (0-100)
  - component_scores: dict (component key -> score)
  - recorded_at: str (ISO timestamp of when the entry was recorded)

Max 20 entries retained; oldest pruned first.
Same data_date re-run overwrites the existing entry.
"""

import json
import os
from datetime import datetime
from typing import Any


def load_history(path: str) -> list[dict]:
    """Load history from JSON file. Returns [] on missing/corrupt file."""
    if not os.path.exists(path):
        return []
    try:
        with open(path) as f:
            content = f.read().strip()
            if not content:
                return []
            data = json.loads(content)
            if isinstance(data, list):
                return data
            return []
    except (json.JSONDecodeError, OSError):
        return []


def append_history(
    path: str,
    composite_score: float,
    component_scores: dict[str, Any],
    data_date: str,
) -> list[dict]:
    """Append or overwrite an entry for *data_date*, then save.

    Returns the updated history list.
    """
    history = load_history(path)

    entry = {
        "data_date": data_date,
        "composite_score": composite_score,
        "component_scores": component_scores,
        "recorded_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    }

    # Overwrite if same data_date exists
    history = [h for h in history if h.get("data_date") != data_date]
    history.append(entry)

    # Sort by data_date ascending
    history.sort(key=lambda h: h.get("data_date", ""))

    # Prune to max 20 entries
    if len(history) > 20:
        history = history[-20:]

    with open(path, "w", encoding="utf-8") as f:
        json.dump(history, f, indent=2, default=str)

    return history


def get_trend_summary(history: list[dict], n: int = 5) -> dict:
    """Compute trend direction from the last *n* entries.

    Returns:
        Dict with delta, direction ("improving"/"deteriorating"/"stable"),
        and the entries used.
    """
    if not history:
        return {"delta": 0, "direction": "stable", "entries": []}

    recent = history[-n:]

    if len(recent) < 2:
        return {
            "delta": 0,
            "direction": "stable",
            "entries": recent,
        }

    first_score = recent[0]["composite_score"]
    last_score = recent[-1]["composite_score"]
    delta = round(last_score - first_score, 1)

    if delta > 2:
        direction = "improving"
    elif delta < -2:
        direction = "deteriorating"
    else:
        direction = "stable"

    return {
        "delta": delta,
        "direction": direction,
        "entries": recent,
    }

#!/usr/bin/env python3
"""
Market Breadth Analyzer - Main Orchestrator

Quantifies market breadth health using TraderMonty's public CSV data.
Generates a 0-100 composite score across 6 components.
No API key required.

Usage:
    # Default (uses public CSV URLs):
    python3 market_breadth_analyzer.py

    # Custom URLs:
    python3 market_breadth_analyzer.py \\
        --detail-url "https://example.com/data.csv" \\
        --summary-url "https://example.com/summary.csv"

    # Custom output directory:
    python3 market_breadth_analyzer.py --output-dir ./reports

Output:
    - JSON: market_breadth_YYYY-MM-DD_HHMMSS.json
    - Markdown: market_breadth_YYYY-MM-DD_HHMMSS.md
"""

import argparse
import os
import sys
from datetime import datetime, timezone
from uuid import uuid4

# Add parent directory to path for imports
sys.path.insert(0, os.path.dirname(__file__))

from calculators.bearish_signal_calculator import calculate_bearish_signal
from calculators.cycle_calculator import calculate_cycle_position
from calculators.divergence_calculator import calculate_divergence
from calculators.historical_context_calculator import calculate_historical_percentile
from calculators.ma_crossover_calculator import calculate_ma_crossover
from calculators.trend_level_calculator import calculate_breadth_level_trend
from csv_client import (
    DEFAULT_DETAIL_URL,
    DEFAULT_SUMMARY_URL,
    check_data_freshness,
    fetch_detail_csv,
    fetch_summary_csv,
)
from history_tracker import append_history, get_trend_summary
from provenance_contract import build_provenance, safe_source_url
from report_generator import generate_json_report, generate_markdown_report
from scorer import calculate_composite_score


def parse_arguments():
    parser = argparse.ArgumentParser(
        description="Market Breadth Analyzer - 6-Component Health Scoring"
    )

    parser.add_argument(
        "--detail-url",
        default=DEFAULT_DETAIL_URL,
        help="URL for detail CSV (market_breadth_data.csv)",
    )
    parser.add_argument(
        "--summary-url",
        default=DEFAULT_SUMMARY_URL,
        help="URL for summary CSV (market_breadth_summary.csv)",
    )
    parser.add_argument(
        "--output-dir",
        default=".",
        help="Output directory for reports (default: current directory)",
    )

    parser.add_argument("--run-id", default=None, help="Shared workflow run identifier")
    return parser.parse_args()


def main():
    args = parse_arguments()

    print("=" * 70)
    print("Market Breadth Analyzer")
    print("6-Component Health Scoring (No API Key Required)")
    print("=" * 70)
    print()

    # ========================================================================
    # Step 1: Fetch CSV Data
    # ========================================================================
    print("Step 1: Fetching CSV Data")
    print("-" * 70)

    detail_rows = fetch_detail_csv(args.detail_url)
    if not detail_rows:
        print("ERROR: Cannot proceed without detail CSV data", file=sys.stderr)
        sys.exit(1)

    summary = fetch_summary_csv(args.summary_url)

    provenance = build_provenance(
        provider="trader-monty" if args.detail_url == DEFAULT_DETAIL_URL else "custom-csv",
        endpoint=safe_source_url(args.detail_url),
        retrieved_at=datetime.now(timezone.utc).isoformat(),
        as_of=detail_rows[-1].get("Date"),
        run_id=args.run_id or uuid4().hex,
    )

    provenance["auxiliary_sources"] = [
        {"endpoint": safe_source_url(args.summary_url), "as_of_verified": False}
    ]

    freshness = check_data_freshness(detail_rows)
    if freshness.get("warning"):
        print(f"  WARNING: {freshness['warning']}")
    else:
        print(
            f"  Data freshness: OK "
            f"(latest: {freshness['latest_date']}, {freshness['days_old']} days old)"
        )

    print()

    # ========================================================================
    # Step 2: Calculate Components
    # ========================================================================
    print("Step 2: Calculating Components")
    print("-" * 70)

    # Component 1: Current Breadth Level & Trend (25%)
    print("  [1/6] Current Breadth Level & Trend...", end=" ", flush=True)
    comp1 = calculate_breadth_level_trend(detail_rows)
    print(f"Score: {comp1['score']} ({comp1['signal']})")

    # Component 2: 8MA vs 200MA Crossover (20%)
    print("  [2/6] 8MA vs 200MA Crossover...", end=" ", flush=True)
    comp2 = calculate_ma_crossover(detail_rows)
    print(f"Score: {comp2['score']} ({comp2['signal']})")

    # Component 3: Peak/Trough Cycle Position (20%)
    print("  [3/6] Peak/Trough Cycle Position...", end=" ", flush=True)
    comp3 = calculate_cycle_position(detail_rows)
    print(f"Score: {comp3['score']} ({comp3['signal']})")

    # Component 4: Bearish Signal Status (15%)
    print("  [4/6] Bearish Signal Status...", end=" ", flush=True)
    comp4 = calculate_bearish_signal(detail_rows)
    print(f"Score: {comp4['score']} ({comp4['signal']})")

    # Component 5: Historical Percentile (10%)
    print("  [5/6] Historical Percentile...", end=" ", flush=True)
    comp5 = calculate_historical_percentile(detail_rows, summary)
    print(f"Score: {comp5['score']} ({comp5['signal']})")

    # Component 6: S&P 500 vs Breadth Divergence (10%)
    print("  [6/6] S&P 500 vs Breadth Divergence...", end=" ", flush=True)
    comp6 = calculate_divergence(detail_rows)
    print(f"Score: {comp6['score']} ({comp6['signal']})")

    print()

    # ========================================================================
    # Step 3: Composite Score
    # ========================================================================
    print("Step 3: Calculating Composite Score")
    print("-" * 70)

    component_scores = {
        "breadth_level_trend": comp1["score"],
        "ma_crossover": comp2["score"],
        "cycle_position": comp3["score"],
        "bearish_signal": comp4["score"],
        "historical_percentile": comp5["score"],
        "divergence": comp6["score"],
    }

    data_availability = {
        "breadth_level_trend": comp1.get("data_available", True),
        "ma_crossover": comp2.get("data_available", True),
        "cycle_position": comp3.get("data_available", True),
        "bearish_signal": comp4.get("data_available", True),
        "historical_percentile": comp5.get("data_available", True),
        "divergence": comp6.get("data_available", True),
    }

    composite = calculate_composite_score(component_scores, data_availability)

    print(f"  Composite Score: {composite['composite_score']}/100")
    print(f"  Health Zone: {composite['zone']}")
    print(f"  Equity Exposure: {composite['exposure_guidance']}")
    print(
        f"  Strongest: {composite['strongest_health']['label']} "
        f"({composite['strongest_health']['score']})"
    )
    print(
        f"  Weakest: {composite['weakest_health']['label']} "
        f"({composite['weakest_health']['score']})"
    )
    print()

    # ========================================================================
    # Step 3.5: Score History & Trend
    # ========================================================================
    data_date = detail_rows[-1]["Date"]
    history_file = os.path.join(args.output_dir, "market_breadth_history.json")
    updated_history = append_history(
        history_file,
        composite["composite_score"],
        component_scores,
        data_date,
    )
    trend_summary = get_trend_summary(updated_history)
    if trend_summary["direction"] != "stable" and len(trend_summary["entries"]) >= 2:
        print(
            f"  Score Trend: {trend_summary['direction']} "
            f"(delta {trend_summary['delta']:+.1f} over "
            f"{len(trend_summary['entries'])} observations)"
        )
    print()

    # ========================================================================
    # Step 4: Key Levels
    # ========================================================================
    key_levels = _compute_key_levels(detail_rows, summary)

    # ========================================================================
    # Step 5: Generate Reports
    # ========================================================================
    print("Step 4: Generating Reports")
    print("-" * 70)

    analysis = {
        "data_provenance": provenance,
        "metadata": {
            "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "data_source": "TraderMonty Market Breadth CSV",
            "detail_url": args.detail_url,
            "summary_url": args.summary_url,
            "total_rows": len(detail_rows),
            "data_freshness": freshness,
        },
        "composite": composite,
        "components": {
            "breadth_level_trend": comp1,
            "ma_crossover": comp2,
            "cycle_position": comp3,
            "bearish_signal": comp4,
            "historical_percentile": comp5,
            "divergence": comp6,
        },
        "trend_summary": trend_summary,
        "key_levels": key_levels,
    }

    timestamp = datetime.now().strftime("%Y-%m-%d_%H%M%S")
    json_file = os.path.join(args.output_dir, f"market_breadth_{timestamp}.json")
    md_file = os.path.join(args.output_dir, f"market_breadth_{timestamp}.md")

    generate_json_report(analysis, json_file)
    generate_markdown_report(analysis, md_file)

    print()
    print("=" * 70)
    print("Market Breadth Analysis Complete")
    print("=" * 70)
    print(f"  Composite Score: {composite['composite_score']}/100")
    print(f"  Health Zone: {composite['zone']}")
    print(f"  Equity Exposure: {composite['exposure_guidance']}")
    print(f"  JSON Report: {json_file}")
    print(f"  Markdown Report: {md_file}")
    print()


def _compute_key_levels(rows, summary):
    """Compute key breadth levels to watch."""
    if not rows:
        return {}

    latest = rows[-1]
    latest["Breadth_Index_8MA"]
    ma200 = latest["Breadth_Index_200MA"]

    levels = {}

    # 200MA crossover level
    levels["200MA Level"] = {
        "value": f"{ma200:.4f}",
        "significance": (
            "Key support/resistance for 8MA. "
            "8MA crossing below is an early warning of deterioration, "
            "not a standalone bearish signal."
        ),
    }

    # 0.40 extreme weakness threshold
    levels["Extreme Weakness (0.40)"] = {
        "value": "0.4000",
        "significance": (
            "8MA below 0.40 marks extreme weakness. "
            "Historically, troughs at this level precede significant rallies."
        ),
    }

    # 0.60 healthy threshold
    levels["Healthy Threshold (0.60)"] = {
        "value": "0.6000",
        "significance": (
            "8MA above 0.60 indicates broad participation. "
            "Below 0.60 = selective market, above = inclusive rally."
        ),
    }

    # Average peak from summary
    avg_peak_str = summary.get("Average Peaks (200MA)", "")
    try:
        avg_peak = float(avg_peak_str)
        levels["Historical Avg Peak"] = {
            "value": f"{avg_peak:.3f}",
            "significance": (
                "Average peak level. Approaching this level suggests "
                "breadth may be near a cyclical high."
            ),
        }
    except (ValueError, TypeError):
        pass

    # Average trough from summary
    avg_trough_str = summary.get("Average Troughs (8MA < 0.4)", "")
    try:
        avg_trough = float(avg_trough_str)
        levels["Historical Avg Trough"] = {
            "value": f"{avg_trough:.3f}",
            "significance": (
                "Average extreme trough level. Reaching this level "
                "is a potential contrarian buy signal."
            ),
        }
    except (ValueError, TypeError):
        pass

    return levels


if __name__ == "__main__":
    main()

"""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, "", ""))

#!/usr/bin/env python3
"""
Market Breadth Analyzer - Report Generator

Generates JSON and Markdown reports for market breadth health analysis.
"""

import json


def generate_json_report(analysis: dict, output_file: str):
    """Save full analysis as JSON."""
    with open(output_file, "w", encoding="utf-8") as f:
        json.dump(analysis, f, indent=2, default=str)
    print(f"  JSON report saved to: {output_file}")


def generate_markdown_report(analysis: dict, output_file: str):
    """Generate comprehensive Markdown report."""
    lines = []
    composite = analysis.get("composite", {})
    components = analysis.get("components", {})
    metadata = analysis.get("metadata", {})
    freshness = metadata.get("data_freshness", {})

    score = composite.get("composite_score", 0)
    zone = composite.get("zone", "Unknown")
    exposure = composite.get("exposure_guidance", "N/A")

    # Header
    lines.append("# Market Breadth Analyzer Report")
    lines.append("")
    lines.append(f"**Generated:** {metadata.get('generated_at', 'N/A')}")
    lines.append("**Data Source:** TraderMonty Market Breadth CSV (no API key required)")
    latest = freshness.get("latest_date", "N/A")
    days_old = freshness.get("days_old", "?")
    last_modified = freshness.get("last_modified")
    lm_str = f" | CSV last modified: {last_modified}" if last_modified else ""
    lines.append(f"**Latest Data:** {latest} ({days_old} days old{lm_str})")
    lines.append(
        "**Live Dashboard:** [Interactive Chart]"
        "(https://tradermonty.github.io/market-breadth-analysis/)"
    )
    lines.append("")

    # Freshness warning
    warning = freshness.get("warning")
    if warning:
        lines.append(f"> **WARNING:** {warning}")
        lines.append("")

    # Overall Assessment
    lines.append("---")
    lines.append("")
    lines.append("## Overall Assessment")
    lines.append("")
    zone_emoji = _zone_emoji(composite.get("zone_color", ""))
    lines.append("| Metric | Value |")
    lines.append("|--------|-------|")
    lines.append(f"| **Composite Score** | **{score}/100** |")
    lines.append(f"| **Health Zone** | {zone_emoji} {zone} |")
    lines.append(f"| **Equity Exposure** | {exposure} |")
    lines.append(
        f"| **Strongest** | {composite.get('strongest_health', {}).get('label', 'N/A')} "
        f"({composite.get('strongest_health', {}).get('score', 0)}/100) |"
    )
    lines.append(
        f"| **Weakest** | {composite.get('weakest_health', {}).get('label', 'N/A')} "
        f"({composite.get('weakest_health', {}).get('score', 0)}/100) |"
    )
    dq = composite.get("data_quality", {})
    if dq:
        lines.append(f"| **Data Quality** | {dq.get('label', 'N/A')} |")
    lines.append("")

    # Guidance
    lines.append(f"> **Guidance:** {composite.get('guidance', '')}")
    c1_mod = components.get("breadth_level_trend", {}).get("direction_modifier", 0)
    c2_mod = components.get("ma_crossover", {}).get("direction_modifier", 0)
    if c1_mod < 0 or c2_mod < 0:
        lines.append(
            f"> **Caution:** 8MA is falling "
            f"(C1 modifier: {c1_mod:+d}, C2 modifier: {c2_mod:+d}). "
            f"Monitor for further deterioration."
        )
    lines.append("")

    # Component Scores Table
    lines.append("---")
    lines.append("")
    lines.append("## Component Scores")
    lines.append("")
    lines.append("| # | Component | Weight | Eff. Weight | Score | Contribution | Signal |")
    lines.append("|---|-----------|--------|-------------|-------|--------------|--------|")

    component_order = [
        "breadth_level_trend",
        "ma_crossover",
        "cycle_position",
        "bearish_signal",
        "historical_percentile",
        "divergence",
    ]

    excluded_components = composite.get("excluded_components", [])

    for i, key in enumerate(component_order, 1):
        comp = composite.get("component_scores", {}).get(key, {})
        detail = components.get(key, {})
        signal = detail.get("signal", "N/A")
        score_val = int(round(comp.get("score", 0)))
        weight_pct = f"{comp.get('weight', 0) * 100:.0f}%"
        eff_weight = comp.get("effective_weight", comp.get("weight", 0))
        eff_weight_pct = f"{eff_weight * 100:.0f}%"
        contribution = comp.get("weighted_contribution", 0)
        is_excluded = not comp.get("data_available", True)
        bar = _score_bar(score_val)

        label = comp.get("label", key)
        if is_excluded:
            label += " (excluded)"

        lines.append(
            f"| {i} | **{label}** | {weight_pct} | {eff_weight_pct} | "
            f"{bar} {score_val} | {contribution:.1f} | {signal} |"
        )

    lines.append("")

    if excluded_components:
        lines.append(
            f"> **Note:** {len(excluded_components)} component(s) excluded due to "
            f"insufficient data: {', '.join(excluded_components)}. "
            f"Weights have been redistributed among available components."
        )
        lines.append("")

    # Component Details
    lines.append("---")
    lines.append("")
    lines.append("## Component Details")
    lines.append("")

    # C1: Breadth Level & Trend
    c1 = components.get("breadth_level_trend", {})
    lines.append("### 1. Current Breadth Level & Trend")
    lines.append("")
    if c1:
        lines.append(f"- **8MA Level:** {c1.get('current_8ma', 0):.4f}")
        lines.append(f"- **200MA Level:** {c1.get('current_200ma', 0):.4f}")
        lines.append(f"- **200MA Trend:** {'Uptrend' if c1.get('trend') == 1 else 'Downtrend'}")
        lines.append(f"- **Level Score:** {c1.get('level_score', 'N/A')}")
        lines.append(f"- **Trend Score:** {c1.get('trend_score', 'N/A')}")
        if c1.get("ma8_direction"):
            lines.append(f"- **8MA Direction (5d):** {c1['ma8_direction']}")
            modifier = c1.get("direction_modifier", 0)
            if modifier != 0:
                lines.append(f"- **Direction Modifier:** {modifier:+d}")
    lines.append("")

    # C2: MA Crossover
    c2 = components.get("ma_crossover", {})
    lines.append("### 2. 8MA vs 200MA Crossover Dynamics")
    lines.append("")
    if c2:
        lines.append(f"- **Gap (8MA - 200MA):** {c2.get('gap', 0):+.4f}")
        lines.append(f"- **Gap Score:** {c2.get('gap_score', 'N/A')}")
        lines.append(f"- **8MA Direction (5d):** {c2.get('ma8_direction', 'N/A')}")
        lines.append(f"- **Direction Modifier:** {c2.get('direction_modifier', 0):+d}")
    lines.append("")

    # C3: Cycle Position
    c3 = components.get("cycle_position", {})
    lines.append("### 3. Peak/Trough Cycle Position")
    lines.append("")
    if c3:
        lines.append(f"- **Latest Marker:** {c3.get('latest_marker_type', 'None')}")
        lines.append(f"- **Days Since Marker:** {c3.get('days_since_marker', 'N/A')}")
        lines.append(f"- **8MA Trend:** {c3.get('ma8_trend', 'N/A')}")
        if c3.get("extreme_trough"):
            lines.append("- **Extreme Trough (8MA < 0.4):** Yes (+10 bonus)")
    lines.append("")

    # C4: Bearish Signal
    c4 = components.get("bearish_signal", {})
    lines.append("### 4. Bearish Signal Status")
    lines.append("")
    if c4:
        lines.append(f"- **Bearish Signal Active:** {'YES' if c4.get('signal_active') else 'No'}")
        lines.append(f"- **200MA Trend:** {'Uptrend' if c4.get('trend') == 1 else 'Downtrend'}")
        lines.append(f"- **Current 8MA:** {c4.get('current_8ma', 0):.4f}")
        in_pink = c4.get("in_pink_zone", False)
        pink_days = c4.get("pink_zone_days", 0)
        if in_pink:
            lines.append(
                f"- **Pink Zone:** YES ({pink_days} consecutive days) "
                f"- 200MA downtrend + 8MA below 200MA"
            )
        else:
            lines.append("- **Pink Zone:** No (outside bearish region)")
        lines.append(f"- **Base Score:** {c4.get('base_score', 'N/A')}")
        lines.append(f"- **Context Adjustment:** {c4.get('context_adjustment', 0):+d}")
        pink_adj = c4.get("pink_zone_adjustment", 0)
        if pink_adj != 0:
            lines.append(f"- **Pink Zone Adjustment:** {pink_adj:+d}")
    lines.append("")

    # C5: Historical Percentile
    c5 = components.get("historical_percentile", {})
    lines.append("### 5. Historical Percentile")
    lines.append("")
    if c5:
        lines.append(f"- **Current 8MA:** {c5.get('current_8ma', 0):.4f}")
        lines.append(f"- **Percentile Rank:** {c5.get('percentile_rank', 'N/A'):.1f}%")
        lines.append(f"- **Avg Peak (200MA):** {c5.get('avg_peak', 'N/A')}")
        lines.append(f"- **Avg Trough (8MA<0.4):** {c5.get('avg_trough', 'N/A')}")
        adj = c5.get("adjustment", 0)
        if adj != 0:
            label = "Overheated (-10)" if adj < 0 else "Oversold (+10)"
            lines.append(f"- **Adjustment:** {label}")
    lines.append("")

    # C6: Divergence
    c6 = components.get("divergence", {})
    lines.append("### 6. S&P 500 vs Breadth Divergence")
    lines.append("")
    if c6:
        lines.append(f"- **S&P 500 60d Change:** {c6.get('sp500_pct_change', 0):+.2f}%")
        lines.append(f"- **Breadth 8MA 60d Change:** {c6.get('breadth_change', 0):+.4f}")
        lines.append(f"- **Divergence Type:** {c6.get('divergence_type', 'N/A')}")

        # Dual-window details (new)
        w60 = c6.get("window_60d")
        w20 = c6.get("window_20d")
        if w60 and w20:
            lines.append("")
            lines.append(
                f"- **60-Day Window:** Score {w60['score']}, "
                f"{w60['divergence_type']} "
                f"(lookback: {w60.get('lookback_days', 60)}d)"
            )
            lines.append(
                f"- **20-Day Window:** Score {w20['score']}, "
                f"{w20['divergence_type']} "
                f"(lookback: {w20.get('lookback_days', 20)}d)"
            )

        if c6.get("early_warning"):
            lines.append("")
            lines.append(
                "> **Early Warning:** Short-term (20d) bearish divergence "
                "detected while long-term (60d) trend remains healthy. "
                "Monitor for potential deterioration."
            )
    lines.append("")

    # Score Trend (show N/A for single entry, full table for 2+)
    trend_summary = analysis.get("trend_summary")
    if trend_summary and len(trend_summary.get("entries", [])) >= 1:
        entries = trend_summary["entries"]
        lines.append("---")
        lines.append("")
        lines.append("## Score Trend")
        lines.append("")
        if len(entries) == 1:
            lines.append("**Direction:** N/A (single observation — trend requires 2+ data points)")
        else:
            direction = trend_summary["direction"].capitalize()
            delta = trend_summary["delta"]
            lines.append(
                f"**Direction:** {direction} ({delta:+.1f} over {len(entries)} observations)"
            )
            lines.append("")
            lines.append("| Date | Score |")
            lines.append("|------|-------|")
            for entry in entries:
                lines.append(f"| {entry['data_date']} | {entry['composite_score']:.1f} |")
        lines.append("")

    # Key Levels to Watch
    lines.append("---")
    lines.append("")
    lines.append("## Key Levels to Watch")
    lines.append("")
    key_levels = analysis.get("key_levels", {})
    if key_levels:
        lines.append("| Level | Value | Significance |")
        lines.append("|-------|-------|-------------|")
        for level_name, level_info in key_levels.items():
            lines.append(
                f"| {level_name} | {level_info.get('value', 'N/A')} | "
                f"{level_info.get('significance', '')} |"
            )
    else:
        lines.append("*No key levels computed.*")
    lines.append("")

    # Recommended Actions
    lines.append("---")
    lines.append("")
    lines.append("## Recommended Actions")
    lines.append("")
    lines.append(f"**Health Zone:** {zone}")
    lines.append(f"**Equity Exposure:** {exposure}")
    lines.append("")
    for action in composite.get("actions", []):
        lines.append(f"- {action}")
    # Inject cautionary actions when 8MA is falling
    c1_mod_act = components.get("breadth_level_trend", {}).get("direction_modifier", 0)
    c2_mod_act = components.get("ma_crossover", {}).get("direction_modifier", 0)
    if c1_mod_act < 0 or c2_mod_act < 0:
        lines.append("- Reduce new position sizes until 8MA stabilizes")
        lines.append("- Tighten stop-loss levels on existing positions")
    lines.append("")

    # Methodology
    lines.append("---")
    lines.append("")
    lines.append("## Methodology")
    lines.append("")
    lines.append(
        "This analysis uses TraderMonty's market breadth dataset to quantify "
        "market participation health across 6 dimensions:"
    )
    lines.append("")
    lines.append("1. **Breadth Level & Trend (25%):** Current 8MA level and 200MA trend direction")
    lines.append("2. **MA Crossover (20%):** Gap between 8MA and 200MA with momentum detection")
    lines.append("3. **Cycle Position (20%):** Position relative to recent peaks/troughs")
    lines.append("4. **Bearish Signal (15%):** Backtested bearish signal flag from dataset")
    lines.append("5. **Historical Percentile (10%):** Current level vs full history distribution")
    lines.append("6. **Divergence (10%):** S&P 500 price vs breadth directional agreement")
    lines.append("")
    lines.append(
        "Composite score: 0-100 (100 = maximum health). "
        "No API key required - uses freely available CSV data."
    )
    lines.append("")
    lines.append("For detailed methodology, see `references/breadth_analysis_methodology.md`.")
    lines.append("")

    # Disclaimer
    lines.append("---")
    lines.append("")
    lines.append(
        "**Disclaimer:** This analysis is for educational and informational purposes only. "
        "Not investment advice. Past patterns may not predict future outcomes. "
        "Conduct your own research and consult a financial advisor before making "
        "investment decisions."
    )
    lines.append("")

    with open(output_file, "w", encoding="utf-8") as f:
        f.write("\n".join(lines))

    print(f"  Markdown report saved to: {output_file}")


def _zone_emoji(color: str) -> str:
    mapping = {
        "green": "🟢",
        "blue": "🔵",
        "yellow": "🟡",
        "orange": "🟠",
        "red": "🔴",
    }
    return mapping.get(color, "⚪")


def _score_bar(score: int) -> str:
    """Text bar for score visualization (100 = full/healthy)."""
    if score >= 80:
        return "████"
    elif score >= 60:
        return "███░"
    elif score >= 40:
        return "██░░"
    elif score >= 20:
        return "█░░░"
    else:
        return "░░░░"

#!/usr/bin/env python3
"""
Market Breadth Analyzer - Composite Scoring Engine

Combines 6 component scores into a weighted composite (0-100).
Score direction: 100 = Healthy, 0 = Critical (breadth is a health indicator).

Component Weights:
1. Current Breadth Level & Trend:  25%
2. 8MA vs 200MA Crossover:        20%
3. Peak/Trough Cycle Position:    20%
4. Bearish Signal Status:         15%
5. Historical Percentile:         10%
6. S&P 500 vs Breadth Divergence: 10%
Total: 100%

When a component has data_available=False, its weight is proportionally
redistributed among the remaining available components.

Health Zone Mapping (100 = Healthy):
  80-100: Strong    - Full equity exposure
  60-79:  Healthy   - Normal operations
  40-59:  Neutral   - Selective positioning
  20-39:  Weakening - Profit-taking, raise cash
  0-19:   Critical  - Capital preservation
"""

from typing import Optional

COMPONENT_WEIGHTS = {
    "breadth_level_trend": 0.25,
    "ma_crossover": 0.20,
    "cycle_position": 0.20,
    "bearish_signal": 0.15,
    "historical_percentile": 0.10,
    "divergence": 0.10,
}

COMPONENT_LABELS = {
    "breadth_level_trend": "Current Breadth Level & Trend",
    "ma_crossover": "8MA vs 200MA Crossover",
    "cycle_position": "Peak/Trough Cycle Position",
    "bearish_signal": "Bearish Signal Status",
    "historical_percentile": "Historical Percentile",
    "divergence": "S&P 500 vs Breadth Divergence",
}


def calculate_composite_score(
    component_scores: dict[str, float],
    data_availability: Optional[dict[str, bool]] = None,
) -> dict:
    """
    Calculate weighted composite market breadth health score.

    Args:
        component_scores: Dict with keys matching COMPONENT_WEIGHTS, each 0-100.
        data_availability: Optional dict mapping component key -> bool.
            When a component is False, its weight is redistributed proportionally.

    Returns:
        Dict with composite_score, zone, exposure_guidance, guidance,
        strongest/weakest components, component breakdown, and data_quality.
    """
    if data_availability is None:
        data_availability = {}

    # Determine which components are available
    available = {k: data_availability.get(k, True) for k in COMPONENT_WEIGHTS}

    # Calculate sum of available weights for redistribution
    available_total_weight = sum(w for k, w in COMPONENT_WEIGHTS.items() if available[k])

    # Build effective weights
    effective_weights = {}
    for key, weight in COMPONENT_WEIGHTS.items():
        if not available[key]:
            effective_weights[key] = 0.0
        elif available_total_weight > 0:
            effective_weights[key] = weight / available_total_weight
        else:
            effective_weights[key] = 0.0

    # Excluded components
    excluded_components = [COMPONENT_LABELS[k] for k in COMPONENT_WEIGHTS if not available[k]]

    # Zero-division guard: all components unavailable
    if available_total_weight == 0:
        composite = 50.0
        zone_info = _interpret_zone(composite)
        # Override zone to Neutral with warning
        zone_info["zone"] = "Neutral"
        zone_info["guidance"] = (
            "All component data is unavailable. Score is a reference value only."
        )
    else:
        # Weighted composite using effective weights
        composite = 0.0
        for key in COMPONENT_WEIGHTS:
            score = component_scores.get(key, 50)
            composite += score * effective_weights[key]
        composite = round(composite, 1)
        zone_info = _interpret_zone(composite)

    # Identify strongest and weakest among AVAILABLE components only
    valid_scores = {
        k: v
        for k, v in component_scores.items()
        if k in COMPONENT_WEIGHTS and available.get(k, True)
    }

    if valid_scores:
        strongest_health = max(valid_scores, key=valid_scores.get)
        weakest_health = min(valid_scores, key=valid_scores.get)
    else:
        strongest_health = "N/A"
        weakest_health = "N/A"

    # Data quality
    available_count = sum(1 for k in COMPONENT_WEIGHTS if available[k])
    total_components = len(COMPONENT_WEIGHTS)
    missing_components = [COMPONENT_LABELS[k] for k in COMPONENT_WEIGHTS if not available[k]]

    if available_count == total_components:
        quality_label = f"Complete ({available_count}/{total_components} components)"
    elif available_count >= total_components - 2:
        quality_label = (
            f"Partial ({available_count}/{total_components} components) - interpret with caution"
        )
    else:
        quality_label = (
            f"Limited ({available_count}/{total_components} components) - low confidence"
        )

    data_quality = {
        "available_count": available_count,
        "total_components": total_components,
        "label": quality_label,
        "missing_components": missing_components,
    }

    return {
        "composite_score": composite,
        "zone": zone_info["zone"],
        "zone_color": zone_info["color"],
        "exposure_guidance": zone_info["exposure_guidance"],
        "guidance": zone_info["guidance"],
        "actions": zone_info["actions"],
        "strongest_health": {
            "component": strongest_health,
            "label": COMPONENT_LABELS.get(strongest_health, strongest_health),
            "score": valid_scores.get(strongest_health, 0),
        },
        "weakest_health": {
            "component": weakest_health,
            "label": COMPONENT_LABELS.get(weakest_health, weakest_health),
            "score": valid_scores.get(weakest_health, 0),
        },
        "excluded_components": excluded_components,
        "data_quality": data_quality,
        "component_scores": {
            k: {
                "score": component_scores.get(k, 50),
                "weight": w,
                "effective_weight": round(effective_weights[k], 6),
                "weighted_contribution": round(
                    component_scores.get(k, 50) * effective_weights[k], 1
                ),
                "data_available": available[k],
                "label": COMPONENT_LABELS[k],
            }
            for k, w in COMPONENT_WEIGHTS.items()
        },
    }


def _interpret_zone(composite: float) -> dict:
    """Map composite score to health zone (100 = healthy)."""
    if composite >= 80:
        return {
            "zone": "Strong",
            "color": "green",
            "exposure_guidance": "90-100%",
            "guidance": ("Broad market participation. Maintain full equity exposure."),
            "actions": [
                "Full position sizing allowed",
                "New entries on pullbacks encouraged",
                "Wide stop-losses acceptable",
                "Growth and momentum strategies favored",
            ],
        }
    elif composite >= 60:
        return {
            "zone": "Healthy",
            "color": "blue",
            "exposure_guidance": "75-90%",
            "guidance": ("Above-average breadth. Normal operations with standard risk management."),
            "actions": [
                "Normal position sizing",
                "Standard stop-loss levels",
                "New position entries allowed",
                "Monitor for deterioration in leading indicators",
            ],
        }
    elif composite >= 40:
        return {
            "zone": "Neutral",
            "color": "yellow",
            "exposure_guidance": "60-75%",
            "guidance": (
                "Mixed signals. Be selective with new positions and tighten risk controls."
            ),
            "actions": [
                "Reduce new position sizes by 25-50%",
                "Tighten stop-losses",
                "Focus on stocks with strong relative strength",
                "Avoid lagging sectors and speculative names",
            ],
        }
    elif composite >= 20:
        return {
            "zone": "Weakening",
            "color": "orange",
            "exposure_guidance": "40-60%",
            "guidance": ("Breadth deteriorating. Begin profit-taking and raise cash allocation."),
            "actions": [
                "Take profits on weakest 25-40% of positions",
                "No new momentum entries",
                "Raise cash allocation significantly",
                "Consider defensive sector rotation (XLU, XLP, XLV)",
                "Watch for cycle trough signal for re-entry",
            ],
        }
    else:
        return {
            "zone": "Critical",
            "color": "red",
            "exposure_guidance": "25-40%",
            "guidance": (
                "Severe breadth weakness. Capital preservation is the priority. "
                "Watch for trough formation as re-entry signal."
            ),
            "actions": [
                "Maximum cash allocation (60-75%)",
                "Only hold strongest relative strength leaders",
                "Consider hedges (put options, inverse ETFs)",
                "Monitor for extreme trough (8MA < 0.4) as contrarian buy signal",
                "Prepare watchlist for recovery: quality stocks near support",
            ],
        }

"""Shared fixtures for market-breadth-analyzer tests."""

import os
import sys

import pytest

# Ensure scripts/ is on sys.path so calculators and scorer can be imported.
SCRIPTS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if SCRIPTS_DIR not in sys.path:
    sys.path.insert(0, SCRIPTS_DIR)


@pytest.fixture
def make_row():
    """Return a factory that creates a single mock detail row.

    Usage::

        row = make_row(Date="2025-01-10", Breadth_Index_8MA=0.55)
    """

    def _factory(**overrides):
        defaults = {
            "Date": "2025-01-01",
            "S&P500_Price": 5000.0,
            "Breadth_Index_Raw": 0.50,
            "Breadth_Index_200MA": 0.50,
            "Breadth_Index_8MA": 0.50,
            "Breadth_200MA_Trend": 1,
            "Bearish_Signal": False,
            "Is_Peak": False,
            "Is_Trough": False,
            "Is_Trough_8MA_Below_04": False,
        }
        defaults.update(overrides)
        return defaults

    return _factory


@pytest.fixture
def make_rows(make_row):
    """Return a factory that creates *n* rows with sequential dates.

    Usage::

        rows = make_rows(10, Breadth_Index_8MA=0.55)

    The ``Date`` field is auto-generated as ``2025-01-01``, ``2025-01-02``, ...
    Individual overrides per-row are not supported; use ``make_row`` directly.
    """

    def _factory(n, **overrides):
        rows = []
        for i in range(n):
            day = f"2025-01-{i + 1:02d}"
            rows.append(make_row(Date=day, **overrides))
        return rows

    return _factory

"""Tests for Component 4: Bearish Signal Calculator."""

from calculators.bearish_signal_calculator import calculate_bearish_signal


class TestEmptyData:
    def test_empty_rows_returns_unavailable(self):
        result = calculate_bearish_signal([])
        assert result["score"] == 50
        assert result["data_available"] is False


class TestBaseScore:
    """Base score matrix: signal x trend."""

    def test_signal_off_trend_up(self, make_row):
        rows = [make_row(Bearish_Signal=False, Breadth_200MA_Trend=1)]
        result = calculate_bearish_signal(rows)
        assert result["base_score"] == 85
        assert result["score"] == 85

    def test_signal_off_trend_down(self, make_row):
        rows = [
            make_row(
                Bearish_Signal=False,
                Breadth_200MA_Trend=-1,
                Breadth_Index_8MA=0.55,
                Breadth_Index_200MA=0.50,
            )
        ]
        result = calculate_bearish_signal(rows)
        assert result["base_score"] == 50

    def test_signal_on_trend_up(self, make_row):
        rows = [make_row(Bearish_Signal=True, Breadth_200MA_Trend=1, Breadth_Index_8MA=0.40)]
        result = calculate_bearish_signal(rows)
        assert result["base_score"] == 30

    def test_signal_on_trend_down(self, make_row):
        rows = [make_row(Bearish_Signal=True, Breadth_200MA_Trend=-1, Breadth_Index_8MA=0.40)]
        result = calculate_bearish_signal(rows)
        assert result["base_score"] == 10


class TestContextAdjustment:
    """Context adjustment when signal is active."""

    def test_signal_on_high_8ma_gets_plus15(self, make_row):
        rows = [make_row(Bearish_Signal=True, Breadth_200MA_Trend=1, Breadth_Index_8MA=0.55)]
        result = calculate_bearish_signal(rows)
        assert result["context_adjustment"] == 15
        assert result["score"] == 30 + 15  # base + adj

    def test_signal_on_low_8ma_gets_minus5(self, make_row):
        rows = [make_row(Bearish_Signal=True, Breadth_200MA_Trend=-1, Breadth_Index_8MA=0.20)]
        result = calculate_bearish_signal(rows)
        assert result["context_adjustment"] == -5
        assert result["score"] == 10 - 5  # base + adj

    def test_no_adjustment_when_signal_off(self, make_row):
        rows = [make_row(Bearish_Signal=False, Breadth_200MA_Trend=1, Breadth_Index_8MA=0.55)]
        result = calculate_bearish_signal(rows)
        assert result["context_adjustment"] == 0


class TestPinkZone:
    """Pink Zone = trend=-1 AND 8MA < 200MA."""

    def test_pink_zone_detected(self, make_row):
        rows = [
            make_row(
                Breadth_200MA_Trend=-1,
                Breadth_Index_8MA=0.45,
                Breadth_Index_200MA=0.50,
                Bearish_Signal=False,
            )
        ]
        result = calculate_bearish_signal(rows)
        assert result["in_pink_zone"] is True

    def test_pink_zone_penalty_without_signal(self, make_row):
        rows = [
            make_row(
                Breadth_200MA_Trend=-1,
                Breadth_Index_8MA=0.45,
                Breadth_Index_200MA=0.50,
                Bearish_Signal=False,
            )
        ]
        result = calculate_bearish_signal(rows)
        assert result["pink_zone_adjustment"] == -10
        assert result["score"] == 50 - 10  # base=50 (off+down) - 10

    def test_no_pink_zone_when_8ma_above_200ma(self, make_row):
        rows = [
            make_row(
                Breadth_200MA_Trend=-1,
                Breadth_Index_8MA=0.55,
                Breadth_Index_200MA=0.50,
                Bearish_Signal=False,
            )
        ]
        result = calculate_bearish_signal(rows)
        assert result["in_pink_zone"] is False
        assert result["pink_zone_adjustment"] == 0

    def test_no_pink_zone_penalty_when_signal_active(self, make_row):
        rows = [
            make_row(
                Breadth_200MA_Trend=-1,
                Breadth_Index_8MA=0.45,
                Breadth_Index_200MA=0.50,
                Bearish_Signal=True,
            )
        ]
        result = calculate_bearish_signal(rows)
        # Pink zone detected but no extra penalty when signal already active
        assert result["pink_zone_adjustment"] == 0

    def test_pink_zone_days_counted(self, make_row):
        rows = [
            make_row(
                Date="2025-01-01",
                Breadth_200MA_Trend=1,
                Breadth_Index_8MA=0.55,
                Breadth_Index_200MA=0.50,
            ),
            make_row(
                Date="2025-01-02",
                Breadth_200MA_Trend=-1,
                Breadth_Index_8MA=0.48,
                Breadth_Index_200MA=0.50,
                Bearish_Signal=False,
            ),
            make_row(
                Date="2025-01-03",
                Breadth_200MA_Trend=-1,
                Breadth_Index_8MA=0.46,
                Breadth_Index_200MA=0.50,
                Bearish_Signal=False,
            ),
        ]
        result = calculate_bearish_signal(rows)
        assert result["pink_zone_days"] == 2


class TestScoreClamping:
    """Score should be clamped to 0-100."""

    def test_score_does_not_exceed_100(self, make_row):
        rows = [make_row(Bearish_Signal=False, Breadth_200MA_Trend=1)]
        result = calculate_bearish_signal(rows)
        assert 0 <= result["score"] <= 100

    def test_score_does_not_go_below_0(self, make_row):
        rows = [make_row(Bearish_Signal=True, Breadth_200MA_Trend=-1, Breadth_Index_8MA=0.20)]
        result = calculate_bearish_signal(rows)
        assert result["score"] >= 0

"""Behavioral tests for the market breadth CSV client."""

from datetime import datetime
from unittest.mock import Mock

import csv_client
import pytest
import requests

DETAIL_HEADER = ",".join(csv_client.DETAIL_COLUMNS)


def _response(text="", *, headers=None):
    response = Mock()
    response.text = text
    response.headers = headers or {}
    return response


def _detail_row(date, *, price="5000", bearish="false", peak="0"):
    return ",".join(
        [
            date,
            price,
            "0.50",
            "0.45",
            "0.55",
            "1",
            bearish,
            peak,
            "no",
            "yes",
        ]
    )


def test_fetch_detail_csv_parses_types_and_sorts_rows(monkeypatch):
    response = _response(
        "\n".join(
            [
                DETAIL_HEADER,
                _detail_row("2025-01-02", bearish="yes"),
                _detail_row("2025-01-01", price="4990.5", peak="true"),
            ]
        )
    )
    monkeypatch.setattr(csv_client.requests, "get", Mock(return_value=response))

    rows = csv_client.fetch_detail_csv("https://example.test/detail.csv")

    assert [row["Date"] for row in rows] == ["2025-01-01", "2025-01-02"]
    assert rows[0]["S&P500_Price"] == 4990.5
    assert rows[0]["Breadth_200MA_Trend"] == 1
    assert rows[0]["Is_Peak"] is True
    assert rows[0]["Is_Trough"] is False
    assert rows[1]["Bearish_Signal"] is True
    response.raise_for_status.assert_called_once_with()


def test_fetch_detail_csv_skips_bad_rows_and_reports_line(monkeypatch, capsys):
    response = _response(
        "\n".join(
            [
                DETAIL_HEADER,
                _detail_row("2025-01-01", price="not-a-number"),
                _detail_row("2025-01-02"),
            ]
        )
    )
    monkeypatch.setattr(csv_client.requests, "get", Mock(return_value=response))

    rows = csv_client.fetch_detail_csv("https://example.test/detail.csv")

    assert [row["Date"] for row in rows] == ["2025-01-02"]
    assert "WARN: Skipping line 2" in capsys.readouterr().err


@pytest.mark.parametrize(
    "body",
    [
        DETAIL_HEADER,
        "\n".join([DETAIL_HEADER, _detail_row("2025-01-01", price="bad")]),
        "\n".join([DETAIL_HEADER, "2025-01-01"]),
    ],
)
def test_fetch_detail_csv_returns_empty_when_no_valid_rows(monkeypatch, capsys, body):
    monkeypatch.setattr(csv_client.requests, "get", Mock(return_value=_response(body)))

    assert csv_client.fetch_detail_csv("https://example.test/detail.csv") == []
    assert "OK (0 rows)" in capsys.readouterr().out


@pytest.mark.parametrize(
    ("column", "value", "warning"),
    [
        ("Date", "2025-02-30", "invalid Date"),
        ("Date", "2026-1-5", "invalid Date"),
        ("S&P500_Price", "nan", "must be finite"),
        ("Breadth_Index_Raw", "inf", "must be finite"),
        ("Breadth_Index_200MA", "-inf", "must be finite"),
        ("Bearish_Signal", "unknown", "invalid boolean value"),
    ],
)
def test_fetch_detail_csv_rejects_semantically_invalid_rows(
    monkeypatch,
    capsys,
    column,
    value,
    warning,
):
    values = _detail_row("2025-01-01").split(",")
    values[list(csv_client.DETAIL_COLUMNS).index(column)] = value
    body = "\n".join([DETAIL_HEADER, ",".join(values)])
    monkeypatch.setattr(csv_client.requests, "get", Mock(return_value=_response(body)))

    assert csv_client.fetch_detail_csv("https://example.test/detail.csv") == []
    captured = capsys.readouterr()
    assert warning in captured.err
    assert "OK (0 rows)" in captured.out


def test_fetch_detail_csv_transport_error_fails_closed(monkeypatch, capsys):
    monkeypatch.setattr(
        csv_client.requests,
        "get",
        Mock(side_effect=requests.RequestException("offline")),
    )

    assert csv_client.fetch_detail_csv("https://example.test/detail.csv") == []
    assert "Failed to fetch detail CSV: offline" in capsys.readouterr().err


def test_fetch_summary_csv_maps_metrics_and_skips_blank_names(monkeypatch):
    response = _response("Metric,Value\nAverage Peak,0.75\n,ignored\nCurrent,0.55\n")
    monkeypatch.setattr(csv_client.requests, "get", Mock(return_value=response))

    assert csv_client.fetch_summary_csv("https://example.test/summary.csv") == {
        "Average Peak": "0.75",
        "Current": "0.55",
    }
    response.raise_for_status.assert_called_once_with()


def test_fetch_summary_csv_skips_truncated_rows(monkeypatch, capsys):
    response = _response("Metric,Value\nBroken\nCurrent,0.55\n")
    monkeypatch.setattr(csv_client.requests, "get", Mock(return_value=response))

    assert csv_client.fetch_summary_csv("https://example.test/summary.csv") == {"Current": "0.55"}
    assert "WARN: Skipping summary line 2: truncated row" in capsys.readouterr().err


def test_fetch_summary_csv_transport_error_returns_empty(monkeypatch, capsys):
    monkeypatch.setattr(
        csv_client.requests,
        "get",
        Mock(side_effect=requests.RequestException("offline")),
    )

    assert csv_client.fetch_summary_csv("https://example.test/summary.csv") == {}
    assert "Failed to fetch summary CSV: offline" in capsys.readouterr().err


def test_check_data_freshness_handles_missing_and_invalid_dates():
    assert csv_client.check_data_freshness([]) == {
        "is_fresh": False,
        "latest_date": None,
        "days_old": None,
        "last_modified": None,
        "warning": "No data available",
    }

    result = csv_client.check_data_freshness([{"Date": "not-a-date"}])
    assert result["is_fresh"] is False
    assert result["days_old"] is None
    assert result["warning"] == "Cannot parse latest date: not-a-date"


@pytest.mark.parametrize(
    ("latest_date", "max_stale_days", "expected_fresh", "expected_days"),
    [
        ("2025-01-08", 5, True, 2),
        ("2025-01-01", 5, False, 9),
    ],
)
def test_check_data_freshness_is_deterministic(
    monkeypatch,
    latest_date,
    max_stale_days,
    expected_fresh,
    expected_days,
):
    class FixedDateTime(datetime):
        @classmethod
        def now(cls, tz=None):
            return cls(2025, 1, 10)

    monkeypatch.setattr(csv_client, "datetime", FixedDateTime)
    monkeypatch.setattr(csv_client, "_check_last_modified", Mock(return_value="2025-01-09"))

    result = csv_client.check_data_freshness(
        [{"Date": latest_date}],
        max_stale_days=max_stale_days,
        detail_url="https://example.test/detail.csv",
    )

    assert result["is_fresh"] is expected_fresh
    assert result["days_old"] == expected_days
    assert result["last_modified"] == "2025-01-09"
    assert (result["warning"] is None) is expected_fresh


def test_check_last_modified_formats_header_and_swallows_head_failure(monkeypatch):
    response = _response(headers={"Last-Modified": "Wed, 08 Jan 2025 12:30:00 GMT"})
    head = Mock(return_value=response)
    monkeypatch.setattr(csv_client.requests, "head", head)

    assert csv_client._check_last_modified("https://example.test/detail.csv") == (
        "2025-01-08 12:30 UTC"
    )
    head.assert_called_once_with(
        "https://example.test/detail.csv", timeout=10, allow_redirects=True
    )

    head.side_effect = requests.RequestException("offline")
    assert csv_client._check_last_modified("https://example.test/detail.csv") is None


@pytest.mark.parametrize("failure", [requests.Timeout("secret-url"), ValueError("secret-header")])
def test_head_expected_failure_warns_without_sensitive_text(monkeypatch, capsys, failure):
    monkeypatch.setattr(csv_client.requests, "head", Mock(side_effect=failure))
    assert csv_client._check_last_modified("https://example.test/detail.csv") is None
    err = capsys.readouterr().err
    assert "degraded" in err.lower() and "secret" not in err


def test_head_programming_error_propagates(monkeypatch):
    monkeypatch.setattr(csv_client.requests, "head", Mock(side_effect=AssertionError))
    with pytest.raises(AssertionError):
        csv_client._check_last_modified("https://example.test/detail.csv")


def test_missing_head_metadata_marks_freshness_degraded(monkeypatch):
    monkeypatch.setattr(csv_client, "_check_last_modified", Mock(return_value=None))
    result = csv_client.check_data_freshness([{"Date": datetime.now().strftime("%Y-%m-%d")}])
    assert result["is_fresh"] is True
    assert "degraded" in result["warning"].lower()

"""Schema-drift hardening tests for the market breadth CSV client (inline CSV, no network)."""

import sys
from unittest.mock import Mock

import csv_client
import market_breadth_analyzer
import pytest

DETAIL_COLUMNS = list(csv_client.DETAIL_COLUMNS)
DETAIL_HEADER = ",".join(DETAIL_COLUMNS)
GOOD_ROW = "2025-01-01,5000,0.50,0.45,0.55,1,false,0,no,yes"
DRIFT_PREFIX = "ERROR: market_breadth_data.csv schema drift: missing columns ["


def _response(text):
    response = Mock()
    response.text = text
    response.headers = {}
    return response


def _patch_get(monkeypatch, text):
    get = Mock(return_value=_response(text))
    monkeypatch.setattr(csv_client.requests, "get", get)
    return get


def test_detail_header_drift_prints_one_error_and_returns_empty(monkeypatch, capsys):
    header = DETAIL_HEADER.replace("Is_Peak", "Is_Peak_Renamed").replace("Date,", "Trade_Date,", 1)
    _patch_get(monkeypatch, "\n".join([header, GOOD_ROW, GOOD_ROW, GOOD_ROW]))

    assert csv_client.fetch_detail_csv("https://example.test/d.csv") == []

    err = capsys.readouterr().err
    assert err.count("schema drift") == 1
    assert f"{DRIFT_PREFIX}Date,Is_Peak]" in err
    assert "Skipping line" not in err


def test_detail_drift_lists_sorted_comma_separated_names(monkeypatch, capsys):
    header = ",".join(c for c in DETAIL_COLUMNS if c not in ("Is_Trough", "Bearish_Signal"))
    _patch_get(monkeypatch, header + "\n")

    assert csv_client.fetch_detail_csv("https://example.test/d.csv") == []

    assert f"{DRIFT_PREFIX}Bearish_Signal,Is_Trough]" in capsys.readouterr().err


@pytest.mark.parametrize("fetch", ["detail", "summary"])
def test_unparsable_csv_reports_drift_instead_of_crashing(monkeypatch, capsys, fetch):
    oversized = "x" * 200_000  # beyond csv.field_size_limit() -> csv.Error
    if fetch == "detail":
        _patch_get(monkeypatch, f'{DETAIL_HEADER}\n"{oversized}",1\n')
        assert csv_client.fetch_detail_csv("https://example.test/d.csv") == []
        name = "market_breadth_data.csv"
    else:
        _patch_get(monkeypatch, f'Metric,Value\n"{oversized}",1\n')
        assert csv_client.fetch_summary_csv("https://example.test/s.csv") == {}
        name = "market_breadth_summary.csv"
    err = capsys.readouterr().err
    assert err.count(f"ERROR: {name} schema drift: unparsable CSV") == 1


def test_detail_bom_header_is_accepted(monkeypatch, capsys):
    _patch_get(monkeypatch, "" + DETAIL_HEADER + "\n" + GOOD_ROW + "\n")

    rows = csv_client.fetch_detail_csv("https://example.test/d.csv")

    assert [r["Date"] for r in rows] == ["2025-01-01"]
    assert "ERROR" not in capsys.readouterr().err


def test_detail_empty_body_is_drift(monkeypatch, capsys):
    _patch_get(monkeypatch, "")

    assert csv_client.fetch_detail_csv("https://example.test/d.csv") == []
    assert "schema drift" in capsys.readouterr().err


def test_detail_happy_path_has_no_error(monkeypatch, capsys):
    _patch_get(monkeypatch, DETAIL_HEADER + "\n" + GOOD_ROW + "\n")

    assert len(csv_client.fetch_detail_csv("https://example.test/d.csv")) == 1
    assert "ERROR" not in capsys.readouterr().err


REQUIRED = ("Average Peaks (200MA)", "Average Troughs (8MA < 0.4)")


def test_summary_missing_metrics_warns_once_each(monkeypatch, capsys):
    _patch_get(monkeypatch, "Metric,Value\nCurrent,0.55\n")

    assert csv_client.fetch_summary_csv("https://example.test/s.csv") == {"Current": "0.55"}

    err = capsys.readouterr().err
    for name in REQUIRED:
        assert err.count(f"WARN: market_breadth_summary.csv missing metric '{name}'") == 1


def test_summary_with_required_metrics_and_bom_has_no_warning(monkeypatch, capsys):
    body = "Metric,Value\nAverage Peaks (200MA),0.6\nAverage Troughs (8MA < 0.4),0.3\n"
    _patch_get(monkeypatch, "" + body)

    result = csv_client.fetch_summary_csv("https://example.test/s.csv")

    assert result == {"Average Peaks (200MA)": "0.6", "Average Troughs (8MA < 0.4)": "0.3"}
    assert capsys.readouterr().err == ""


def test_summary_only_one_missing_metric_warns_for_that_one(monkeypatch, capsys):
    _patch_get(monkeypatch, "Metric,Value\nAverage Peaks (200MA),0.6\n")

    csv_client.fetch_summary_csv("https://example.test/s.csv")

    err = capsys.readouterr().err
    assert "missing metric 'Average Troughs (8MA < 0.4)'" in err
    assert "missing metric 'Average Peaks (200MA)'" not in err


def test_summary_header_drift_errors_and_returns_empty(monkeypatch, capsys):
    _patch_get(monkeypatch, "Name,Val\nAverage Peaks (200MA),0.6\n")

    assert csv_client.fetch_summary_csv("https://example.test/s.csv") == {}

    err = capsys.readouterr().err
    assert "ERROR: market_breadth_summary.csv schema drift: missing columns [Metric,Value]" in err
    assert "missing metric" not in err


def test_summary_header_drift_single_missing_column(monkeypatch, capsys):
    _patch_get(monkeypatch, "Metric,Val\nx,1\n")

    assert csv_client.fetch_summary_csv("https://example.test/s.csv") == {}
    assert "missing columns [Value]" in capsys.readouterr().err


def test_main_exits_1_for_drifted_detail_csv(monkeypatch, capsys, tmp_path):
    get = _patch_get(monkeypatch, "Wrong,Header\n1,2\n")
    monkeypatch.setattr(sys, "argv", ["market_breadth_analyzer.py", "--output-dir", str(tmp_path)])

    with pytest.raises(SystemExit) as exc:
        market_breadth_analyzer.main()

    assert exc.value.code == 1
    assert capsys.readouterr().err.count("schema drift") == 1
    assert get.call_count == 1
    assert list(tmp_path.iterdir()) == []


def test_detail_padded_header_parses_values(monkeypatch, capsys):
    header = ", ".join(DETAIL_COLUMNS)
    _patch_get(monkeypatch, header + "\n" + GOOD_ROW + "\n")

    rows = csv_client.fetch_detail_csv("https://example.test/d.csv")

    assert len(rows) == 1
    assert rows[0]["Date"] == "2025-01-01"
    assert rows[0]["S&P500_Price"] == 5000.0
    assert "ERROR" not in capsys.readouterr().err


def test_summary_padded_header_parses_values(monkeypatch):
    _patch_get(monkeypatch, "Metric , Value \nCurrent,0.55\n")

    assert csv_client.fetch_summary_csv("https://example.test/s.csv") == {"Current": "0.55"}

"""Tests for Component 3: Peak/Trough Cycle Calculator."""

from calculators.cycle_calculator import calculate_cycle_position


class TestDataAvailability:
    """Marker-not-found should yield data_available=False."""

    def test_no_marker_returns_data_available_false(self, make_rows):
        """120 rows with no peak/trough -> data_available False."""
        rows = make_rows(120)
        result = calculate_cycle_position(rows)
        assert result["data_available"] is False
        assert result["score"] == 50

    def test_marker_present_returns_data_available_true(self, make_rows):
        """A trough marker within lookback -> data_available True."""
        rows = make_rows(120)
        rows[-10]["Is_Trough"] = True
        # Make 8MA rising so we get a proper score
        rows[-1]["Breadth_Index_8MA"] = 0.55
        rows[-6]["Breadth_Index_8MA"] = 0.45
        result = calculate_cycle_position(rows)
        assert result["data_available"] is True
        assert result["score"] != 50  # Should be a real score

    def test_insufficient_data_returns_data_available_false(self, make_rows):
        """Fewer than 10 rows -> data_available False."""
        rows = make_rows(5)
        result = calculate_cycle_position(rows)
        assert result["data_available"] is False

    def test_peak_marker_returns_data_available_true(self, make_rows):
        """A peak marker within lookback -> data_available True."""
        rows = make_rows(30)
        rows[-5]["Is_Peak"] = True
        rows[-1]["Breadth_Index_8MA"] = 0.40
        rows[-6]["Breadth_Index_8MA"] = 0.50
        result = calculate_cycle_position(rows)
        assert result["data_available"] is True

"""Tests for Component 6: Divergence Calculator — dual-window."""

from calculators.divergence_calculator import calculate_divergence


def _make_divergence_rows(make_row, n, sp_start=5000, sp_end=5100, ma8_start=0.50, ma8_end=0.55):
    """Create n rows with linear interpolation of S&P and 8MA."""
    rows = []
    for i in range(n):
        frac = i / max(n - 1, 1)
        sp = sp_start + (sp_end - sp_start) * frac
        ma8 = ma8_start + (ma8_end - ma8_start) * frac
        rows.append(
            make_row(
                **{
                    "S&P500_Price": sp,
                    "Breadth_Index_8MA": ma8,
                    "Date": f"2025-01-{i + 1:02d}",
                }
            )
        )
    return rows


class TestDualWindow:
    """Both windows should be present in output."""

    def test_both_windows_present(self, make_row):
        rows = _make_divergence_rows(make_row, 80)
        result = calculate_divergence(rows)
        assert "window_60d" in result
        assert "window_20d" in result

    def test_composite_score_formula(self, make_row):
        """Composite = 60d * 0.6 + 20d * 0.4."""
        rows = _make_divergence_rows(make_row, 80)
        result = calculate_divergence(rows)
        w60 = result["window_60d"]["score"]
        w20 = result["window_20d"]["score"]
        expected = round(w60 * 0.6 + w20 * 0.4, 1)
        # Allow small rounding tolerance
        assert abs(result["score"] - expected) <= 0.1

    def test_early_warning_flag(self, make_row):
        """20d bearish (<=25) & 60d healthy (>=50) -> early_warning True."""
        # First 60 rows: SP and breadth both rising strongly
        # -> 60d window sees both up -> score=70 (>=50)
        rows = _make_divergence_rows(
            make_row, 60, sp_start=5000, sp_end=5200, ma8_start=0.35, ma8_end=0.65
        )
        # Last 20 rows: SP up >3% but breadth down sharply (<-0.05)
        # -> 20d window score=10 (<=25)
        for i in range(20):
            frac = i / 19
            row = make_row(
                **{
                    "S&P500_Price": 5200 + 200 * frac,
                    "Breadth_Index_8MA": 0.65 - 0.15 * frac,
                    "Date": f"2025-03-{i + 1:02d}",
                }
            )
            rows.append(row)
        result = calculate_divergence(rows)
        assert result["early_warning"] is True

    def test_early_warning_false_when_both_healthy(self, make_row):
        """Both windows healthy -> early_warning False."""
        rows = _make_divergence_rows(
            make_row, 80, sp_start=5000, sp_end=5100, ma8_start=0.50, ma8_end=0.55
        )
        result = calculate_divergence(rows)
        assert result["early_warning"] is False

    def test_backward_compat_top_level_keys(self, make_row):
        """Top-level keys sp500_pct_change, breadth_change, divergence_type
        should still be from the 60d window."""
        rows = _make_divergence_rows(make_row, 80)
        result = calculate_divergence(rows)
        assert "sp500_pct_change" in result
        assert "breadth_change" in result
        assert "divergence_type" in result


class TestShortData:
    """20-59 rows should still produce a valid score."""

    def test_30_rows_uses_shortened_60d_window(self, make_row):
        rows = _make_divergence_rows(make_row, 30)
        result = calculate_divergence(rows)
        assert result["data_available"] is True
        assert result["window_60d"]["lookback_days"] == 30

    def test_20_rows_minimum(self, make_row):
        rows = _make_divergence_rows(make_row, 20)
        result = calculate_divergence(rows)
        assert result["data_available"] is True

    def test_fewer_than_20_unavailable(self, make_row):
        rows = _make_divergence_rows(make_row, 15)
        result = calculate_divergence(rows)
        assert result["data_available"] is False

    def test_60d_lookback_days_correct(self, make_row):
        """With 45 rows, 60d window lookback should be 45."""
        rows = _make_divergence_rows(make_row, 45)
        result = calculate_divergence(rows)
        assert result["window_60d"]["lookback_days"] == 45

    def test_20d_lookback_correct(self, make_row):
        rows = _make_divergence_rows(make_row, 80)
        result = calculate_divergence(rows)
        assert result["window_20d"]["lookback_days"] == 20


class TestNearFlat:
    """Near-flat classification for noise-level movements."""

    def test_near_flat_both_tiny_movements(self):
        from calculators.divergence_calculator import _score_divergence

        score, label = _score_divergence(0.3, 0.005)
        assert "Near-flat" in label

    def test_not_flat_when_sp_significant(self):
        from calculators.divergence_calculator import _score_divergence

        score, label = _score_divergence(1.0, 0.005)
        assert "Near-flat" not in label

    def test_not_flat_when_breadth_significant(self):
        from calculators.divergence_calculator import _score_divergence

        score, label = _score_divergence(0.3, 0.02)
        assert "Near-flat" not in label

    def test_near_flat_both_negative_tiny(self):
        from calculators.divergence_calculator import _score_divergence

        score, label = _score_divergence(-0.2, -0.003)
        assert "Near-flat" in label

    def test_not_flat_when_sp_exactly_at_threshold(self):
        from calculators.divergence_calculator import _score_divergence

        score, label = _score_divergence(0.5, 0.005)
        assert "Near-flat" not in label

"""Tests for Component 5: Historical Context (Percentile) Calculator."""

from calculators.historical_context_calculator import (
    calculate_historical_percentile,
)


def _make_summary(avg_peak="0.729", avg_trough="0.232"):
    """Return a minimal summary dict."""
    return {
        "Average Peaks (200MA)": avg_peak,
        "Average Troughs (8MA < 0.4)": avg_trough,
    }


class TestEmptyData:
    def test_empty_rows_returns_unavailable(self):
        result = calculate_historical_percentile([], _make_summary())
        assert result["score"] == 50
        assert result["data_available"] is False


class TestPercentileScoring:
    """Base score tiers from percentile rank."""

    def test_high_percentile_score_90(self, make_row):
        # 100 rows at 0.40, latest at 0.80 -> percentile near 99
        rows = [make_row(Breadth_Index_8MA=0.40) for _ in range(99)]
        rows.append(make_row(Breadth_Index_8MA=0.80))
        result = calculate_historical_percentile(rows, _make_summary())
        assert result["base_score"] == 90

    def test_mid_percentile_score_50(self, make_row):
        # 50 rows at 0.40, 50 rows at 0.60, latest at 0.50
        rows = [make_row(Breadth_Index_8MA=0.40) for _ in range(50)]
        rows += [make_row(Breadth_Index_8MA=0.60) for _ in range(49)]
        rows.append(make_row(Breadth_Index_8MA=0.50))
        result = calculate_historical_percentile(rows, _make_summary())
        assert result["base_score"] == 50

    def test_low_percentile_score_10(self, make_row):
        # 100 rows at 0.60, latest at 0.20 -> percentile near 0
        rows = [make_row(Breadth_Index_8MA=0.60) for _ in range(99)]
        rows.append(make_row(Breadth_Index_8MA=0.20))
        result = calculate_historical_percentile(rows, _make_summary())
        assert result["base_score"] == 10


class TestExtremeAdjustments:
    """Overheated/oversold adjustments."""

    def test_overheated_adjustment_minus10(self, make_row):
        # current >= avg_peak * 0.95 = 0.729 * 0.95 = 0.6926
        rows = [make_row(Breadth_Index_8MA=0.50) for _ in range(99)]
        rows.append(make_row(Breadth_Index_8MA=0.73))
        result = calculate_historical_percentile(rows, _make_summary())
        assert result["adjustment"] == -10

    def test_oversold_adjustment_plus10(self, make_row):
        # current <= avg_trough * 1.05 = 0.232 * 1.05 = 0.2436
        rows = [make_row(Breadth_Index_8MA=0.50) for _ in range(99)]
        rows.append(make_row(Breadth_Index_8MA=0.23))
        result = calculate_historical_percentile(rows, _make_summary())
        assert result["adjustment"] == 10

    def test_no_adjustment_in_normal_range(self, make_row):
        rows = [make_row(Breadth_Index_8MA=0.50) for _ in range(100)]
        result = calculate_historical_percentile(rows, _make_summary())
        assert result["adjustment"] == 0

    def test_missing_summary_values_no_adjustment(self, make_row):
        rows = [make_row(Breadth_Index_8MA=0.73) for _ in range(100)]
        result = calculate_historical_percentile(rows, {})
        assert result["adjustment"] == 0


class TestOutputFields:
    """Verify all expected output fields are present."""

    def test_output_contains_required_keys(self, make_row):
        rows = [make_row() for _ in range(50)]
        result = calculate_historical_percentile(rows, _make_summary())
        assert result["data_available"] is True
        assert "percentile_rank" in result
        assert "base_score" in result
        assert "avg_peak" in result
        assert "avg_trough" in result
        assert "total_observations" in result
        assert result["total_observations"] == 50

"""Tests for history_tracker.py — score history persistence."""

import json

import pytest
from history_tracker import append_history, get_trend_summary, load_history


@pytest.fixture
def tmp_history(tmp_path):
    """Return a path to a temporary history JSON file."""
    return str(tmp_path / "history.json")


class TestLoadHistory:
    def test_file_not_found(self, tmp_history):
        assert load_history(tmp_history) == []

    def test_empty_file(self, tmp_history):
        with open(tmp_history, "w", encoding="utf-8") as f:
            f.write("")
        assert load_history(tmp_history) == []

    def test_corrupt_file(self, tmp_history):
        with open(tmp_history, "w", encoding="utf-8") as f:
            f.write("{bad json")
        assert load_history(tmp_history) == []

    def test_valid_file(self, tmp_history):
        data = [{"data_date": "2025-01-01", "composite_score": 60.0}]
        with open(tmp_history, "w", encoding="utf-8") as f:
            json.dump(data, f)
        assert load_history(tmp_history) == data


class TestAppendHistory:
    def test_append_creates_file(self, tmp_history):
        append_history(tmp_history, 65.0, {"c1": 70}, "2025-01-10")
        history = load_history(tmp_history)
        assert len(history) == 1
        assert history[0]["composite_score"] == 65.0
        assert history[0]["data_date"] == "2025-01-10"

    def test_append_adds_entry(self, tmp_history):
        append_history(tmp_history, 60.0, {}, "2025-01-09")
        append_history(tmp_history, 65.0, {}, "2025-01-10")
        history = load_history(tmp_history)
        assert len(history) == 2

    def test_same_data_date_overwrites(self, tmp_history):
        append_history(tmp_history, 60.0, {"c1": 70}, "2025-01-10")
        append_history(tmp_history, 65.0, {"c1": 75}, "2025-01-10")
        history = load_history(tmp_history)
        assert len(history) == 1
        assert history[0]["composite_score"] == 65.0

    def test_pruning_at_20(self, tmp_history):
        for i in range(25):
            date = f"2025-01-{i + 1:02d}"
            append_history(tmp_history, 50.0 + i, {}, date)
        history = load_history(tmp_history)
        assert len(history) == 20
        # Oldest entries pruned
        assert history[0]["data_date"] == "2025-01-06"


class TestGetTrendSummary:
    def test_single_entry(self):
        history = [{"data_date": "2025-01-10", "composite_score": 60.0}]
        summary = get_trend_summary(history)
        assert summary["delta"] == 0
        assert summary["direction"] == "stable"
        assert len(summary["entries"]) == 1

    def test_improving_trend(self):
        history = [
            {"data_date": "2025-01-08", "composite_score": 50.0},
            {"data_date": "2025-01-09", "composite_score": 55.0},
            {"data_date": "2025-01-10", "composite_score": 60.0},
        ]
        summary = get_trend_summary(history)
        assert summary["delta"] == 10.0
        assert summary["direction"] == "improving"

    def test_deteriorating_trend(self):
        history = [
            {"data_date": "2025-01-08", "composite_score": 70.0},
            {"data_date": "2025-01-09", "composite_score": 65.0},
            {"data_date": "2025-01-10", "composite_score": 60.0},
        ]
        summary = get_trend_summary(history)
        assert summary["delta"] == -10.0
        assert summary["direction"] == "deteriorating"

    def test_stable_trend(self):
        history = [
            {"data_date": "2025-01-08", "composite_score": 60.0},
            {"data_date": "2025-01-09", "composite_score": 60.5},
            {"data_date": "2025-01-10", "composite_score": 60.0},
        ]
        summary = get_trend_summary(history)
        assert summary["direction"] == "stable"

    def test_n_limits_entries(self):
        history = [
            {"data_date": f"2025-01-{i + 1:02d}", "composite_score": 50.0 + i} for i in range(10)
        ]
        summary = get_trend_summary(history, n=3)
        assert len(summary["entries"]) == 3

    def test_empty_history(self):
        summary = get_trend_summary([])
        assert summary["delta"] == 0
        assert summary["direction"] == "stable"
        assert len(summary["entries"]) == 0

"""Tests for Component 2: MA Crossover Calculator."""

from calculators.ma_crossover_calculator import calculate_ma_crossover


def _make_crossover_rows(make_row, n=10, ma8=0.60, ma200=0.50, ma8_5d_ago=None):
    """Create n rows. The 5th-from-last sets ma8_5d_ago value."""
    if ma8_5d_ago is None:
        ma8_5d_ago = ma8  # no direction change
    rows = []
    for i in range(n):
        if i == n - 6:
            rows.append(
                make_row(
                    Date=f"2025-01-{i + 1:02d}",
                    Breadth_Index_8MA=ma8_5d_ago,
                    Breadth_Index_200MA=ma200,
                )
            )
        else:
            rows.append(
                make_row(
                    Date=f"2025-01-{i + 1:02d}",
                    Breadth_Index_8MA=ma8,
                    Breadth_Index_200MA=ma200,
                )
            )
    return rows


class TestInsufficientData:
    def test_fewer_than_6_rows(self, make_row):
        rows = [make_row() for _ in range(5)]
        result = calculate_ma_crossover(rows)
        assert result["data_available"] is False
        assert result["score"] == 50

    def test_empty_rows(self):
        result = calculate_ma_crossover([])
        assert result["data_available"] is False


class TestGapScoring:
    """Gap = 8MA - 200MA determines base score."""

    def test_large_positive_gap(self, make_row):
        rows = _make_crossover_rows(make_row, ma8=0.70, ma200=0.50)
        result = calculate_ma_crossover(rows)
        assert result["gap_score"] == 95  # gap=0.20 >= 0.15

    def test_moderate_positive_gap(self, make_row):
        rows = _make_crossover_rows(make_row, ma8=0.62, ma200=0.50)
        result = calculate_ma_crossover(rows)
        assert result["gap_score"] == 80  # gap=0.12 >= 0.10

    def test_small_positive_gap(self, make_row):
        rows = _make_crossover_rows(make_row, ma8=0.56, ma200=0.50)
        result = calculate_ma_crossover(rows)
        assert result["gap_score"] == 65  # gap=0.06 >= 0.05

    def test_zero_gap(self, make_row):
        rows = _make_crossover_rows(make_row, ma8=0.50, ma200=0.50)
        result = calculate_ma_crossover(rows)
        assert result["gap_score"] == 50  # gap=0.00

    def test_small_negative_gap(self, make_row):
        rows = _make_crossover_rows(make_row, ma8=0.47, ma200=0.50)
        result = calculate_ma_crossover(rows)
        assert result["gap_score"] == 35  # gap=-0.03 >= -0.05

    def test_large_negative_gap(self, make_row):
        rows = _make_crossover_rows(make_row, ma8=0.35, ma200=0.50)
        result = calculate_ma_crossover(rows)
        assert result["gap_score"] == 5  # gap=-0.15 < -0.10


class TestDirectionModifier:
    """8MA direction modifier based on 5-day lookback."""

    def test_recovery_signal_plus10(self, make_row):
        # Gap < 0 (8MA below 200MA) but 8MA rising -> +10
        rows = _make_crossover_rows(
            make_row,
            ma8=0.48,
            ma200=0.50,
            ma8_5d_ago=0.44,
        )
        result = calculate_ma_crossover(rows)
        assert result["direction_modifier"] == 10
        assert result["ma8_direction"] == "rising"

    def test_deterioration_signal_minus10(self, make_row):
        # Gap > 0 (8MA above 200MA) but 8MA falling -> -10
        rows = _make_crossover_rows(
            make_row,
            ma8=0.55,
            ma200=0.50,
            ma8_5d_ago=0.60,
        )
        result = calculate_ma_crossover(rows)
        assert result["direction_modifier"] == -10
        assert result["ma8_direction"] == "falling"

    def test_no_modifier_gap_positive_rising(self, make_row):
        # Gap > 0 and rising -> no modifier (healthy)
        rows = _make_crossover_rows(
            make_row,
            ma8=0.60,
            ma200=0.50,
            ma8_5d_ago=0.55,
        )
        result = calculate_ma_crossover(rows)
        assert result["direction_modifier"] == 0

    def test_no_modifier_gap_negative_falling(self, make_row):
        # Gap < 0 and falling -> no modifier (already bearish)
        rows = _make_crossover_rows(
            make_row,
            ma8=0.45,
            ma200=0.50,
            ma8_5d_ago=0.48,
        )
        result = calculate_ma_crossover(rows)
        assert result["direction_modifier"] == 0


class TestScoreClamping:
    def test_score_clamped_0_to_100(self, make_row):
        rows = _make_crossover_rows(make_row, ma8=0.70, ma200=0.50)
        result = calculate_ma_crossover(rows)
        assert 0 <= result["score"] <= 100

"""Real orchestration emits provenance from fetched primary rows."""

import json
from datetime import datetime, timedelta
from unittest import mock

import csv_client
import market_breadth_analyzer as analyzer
import pytest


def run_main(output, rows, run_id=None, detail_url=None):
    argv = ["market_breadth_analyzer.py", "--output-dir", str(output)]
    if detail_url is not None:
        argv += ["--detail-url", detail_url]
    if run_id is not None:
        argv += ["--run-id", run_id]
    with (
        mock.patch.object(analyzer, "fetch_detail_csv", return_value=rows),
        mock.patch.object(analyzer, "fetch_summary_csv", return_value={}),
        mock.patch("sys.argv", argv),
        mock.patch.object(
            csv_client.requests,
            "head",
            return_value=mock.Mock(headers={"Last-Modified": "Wed, 08 Jan 2025 12:30:00 GMT"}),
        ) as head,
    ):
        analyzer.main()
    head.assert_called_once()
    assert head.call_args.kwargs == {"timeout": 10, "allow_redirects": True}
    report = next(p for p in output.glob("market_breadth_*.json") if "history" not in p.name)
    return json.loads(report.read_text())["data_provenance"]


def test_main_records_primary_date_and_run(tmp_path, make_rows):
    block = run_main(tmp_path, make_rows(25), "shared-run")
    assert block["run_id"] == "shared-run"
    assert block["as_of"] == "2025-01-25"
    assert datetime.fromisoformat(block["retrieved_at"]).utcoffset() == timedelta(0)
    assert block["provider"] == "trader-monty"
    assert block["endpoint"] == analyzer.DEFAULT_DETAIL_URL
    assert "primary timeseries" in block["scope"]
    assert block["auxiliary_sources"][0]["as_of_verified"] is False
    for flag in ("adjusted", "point_in_time", "survivorship_free", "corporate_actions_handled"):
        assert block[flag] is False


@pytest.mark.parametrize("date", [None, "", "2025-99-99"])
def test_invalid_primary_date_writes_no_report(tmp_path, make_rows, date):
    rows = make_rows(25)
    rows[-1]["Date"] = date
    with pytest.raises(ValueError):
        run_main(tmp_path, rows, "shared-run")
    assert list(tmp_path.iterdir()) == []


def test_default_run_id_is_unique(tmp_path, make_rows):
    (tmp_path / "first").mkdir()
    (tmp_path / "second").mkdir()
    first = run_main(tmp_path / "first", make_rows(25))
    second = run_main(tmp_path / "second", make_rows(25))
    assert first["run_id"] and second["run_id"]
    assert first["run_id"] != second["run_id"]


def test_custom_primary_source_is_identified_honestly(tmp_path, make_rows):
    block = run_main(tmp_path, make_rows(25), detail_url="https://example.com/custom.csv")
    assert block["provider"] == "custom-csv"
    assert block["endpoint"] == "https://example.com/custom.csv"

"""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

"""Tests for report_generator.py — incremental across Steps 1-4."""

import os
import tempfile

from report_generator import generate_markdown_report


def _base_analysis(**overrides):
    """Return a minimal analysis dict for report generation."""
    analysis = {
        "metadata": {
            "generated_at": "2025-01-15 10:00:00",
            "data_source": "Test",
            "detail_url": "http://test",
            "summary_url": "http://test",
            "total_rows": 100,
            "data_freshness": {
                "latest_date": "2025-01-14",
                "days_old": 1,
                "warning": None,
                "last_modified": None,
            },
        },
        "composite": {
            "composite_score": 60.0,
            "zone": "Healthy",
            "zone_color": "blue",
            "exposure_guidance": "75-90%",
            "guidance": "Normal ops",
            "actions": ["Action 1"],
            "strongest_health": {
                "component": "breadth_level_trend",
                "label": "Current Breadth Level & Trend",
                "score": 80,
            },
            "weakest_health": {
                "component": "divergence",
                "label": "S&P 500 vs Breadth Divergence",
                "score": 40,
            },
            "excluded_components": [],
            "data_quality": {
                "available_count": 6,
                "total_components": 6,
                "label": "Complete (6/6 components)",
                "missing_components": [],
            },
            "component_scores": {
                "breadth_level_trend": {
                    "score": 80,
                    "weight": 0.25,
                    "effective_weight": 0.25,
                    "weighted_contribution": 20.0,
                    "data_available": True,
                    "label": "Current Breadth Level & Trend",
                },
                "ma_crossover": {
                    "score": 60,
                    "weight": 0.20,
                    "effective_weight": 0.20,
                    "weighted_contribution": 12.0,
                    "data_available": True,
                    "label": "8MA vs 200MA Crossover",
                },
                "cycle_position": {
                    "score": 50,
                    "weight": 0.20,
                    "effective_weight": 0.20,
                    "weighted_contribution": 10.0,
                    "data_available": True,
                    "label": "Peak/Trough Cycle Position",
                },
                "bearish_signal": {
                    "score": 70,
                    "weight": 0.15,
                    "effective_weight": 0.15,
                    "weighted_contribution": 10.5,
                    "data_available": True,
                    "label": "Bearish Signal Status",
                },
                "historical_percentile": {
                    "score": 55,
                    "weight": 0.10,
                    "effective_weight": 0.10,
                    "weighted_contribution": 5.5,
                    "data_available": True,
                    "label": "Historical Percentile",
                },
                "divergence": {
                    "score": 40,
                    "weight": 0.10,
                    "effective_weight": 0.10,
                    "weighted_contribution": 4.0,
                    "data_available": True,
                    "label": "S&P 500 vs Breadth Divergence",
                },
            },
        },
        "components": {
            "breadth_level_trend": {
                "score": 80,
                "signal": "STRONG: test",
                "data_available": True,
                "current_8ma": 0.65,
                "current_200ma": 0.55,
                "trend": 1,
                "level_score": 80,
                "trend_score": 80,
            },
            "ma_crossover": {
                "score": 60,
                "signal": "POSITIVE: test",
                "data_available": True,
                "gap": 0.10,
                "gap_score": 80,
                "ma8_direction": "rising",
                "direction_modifier": 0,
            },
            "cycle_position": {
                "score": 50,
                "signal": "NEUTRAL: test",
                "data_available": True,
            },
            "bearish_signal": {
                "score": 70,
                "signal": "CLEAR: test",
                "data_available": True,
                "signal_active": False,
                "trend": 1,
                "current_8ma": 0.65,
                "in_pink_zone": False,
                "pink_zone_days": 0,
                "base_score": 85,
                "context_adjustment": -15,
            },
            "historical_percentile": {
                "score": 55,
                "signal": "NORMAL: test",
                "data_available": True,
                "current_8ma": 0.65,
                "percentile_rank": 60.0,
                "avg_peak": 0.73,
                "avg_trough": 0.23,
            },
            "divergence": {
                "score": 40,
                "signal": "Mixed signals: test",
                "data_available": True,
                "sp500_pct_change": 2.5,
                "breadth_change": -0.02,
                "divergence_type": "Mixed signals",
            },
        },
        "key_levels": {},
    }
    analysis.update(overrides)
    return analysis


def _generate_to_string(analysis):
    """Generate markdown report and return as string."""
    with tempfile.NamedTemporaryFile(mode="w", suffix=".md", delete=False) as f:
        tmp = f.name
    try:
        generate_markdown_report(analysis, tmp)
        with open(tmp) as f:
            return f.read()
    finally:
        os.unlink(tmp)


# =========================================================================
# Step 1 tests: excluded components + Eff. Weight
# =========================================================================


class TestStep1ExcludedComponents:
    def test_eff_weight_column_present(self):
        md = _generate_to_string(_base_analysis())
        assert "Eff. Weight" in md

    def test_excluded_component_marked(self):
        analysis = _base_analysis()
        analysis["composite"]["excluded_components"] = ["Peak/Trough Cycle Position"]
        analysis["composite"]["component_scores"]["cycle_position"]["data_available"] = False
        analysis["composite"]["component_scores"]["cycle_position"]["effective_weight"] = 0.0
        md = _generate_to_string(analysis)
        assert "(excluded)" in md

    def test_excluded_note_displayed(self):
        analysis = _base_analysis()
        analysis["composite"]["excluded_components"] = ["Peak/Trough Cycle Position"]
        analysis["composite"]["component_scores"]["cycle_position"]["data_available"] = False
        md = _generate_to_string(analysis)
        assert "excluded due to insufficient data" in md

    def test_no_excluded_no_note(self):
        md = _generate_to_string(_base_analysis())
        assert "excluded due to insufficient data" not in md

    def test_normal_path_layout_intact(self):
        """All-available path should not break existing layout."""
        md = _generate_to_string(_base_analysis())
        assert "## Component Scores" in md
        assert "## Component Details" in md
        assert "## Key Levels to Watch" in md


# =========================================================================
# Step 2 tests: 8MA direction modifier in C1 details
# =========================================================================


class TestStep2DirectionModifier:
    def test_c1_shows_ma8_direction(self):
        analysis = _base_analysis()
        analysis["components"]["breadth_level_trend"]["ma8_direction"] = "rising"
        analysis["components"]["breadth_level_trend"]["direction_modifier"] = 5
        md = _generate_to_string(analysis)
        assert "8MA Direction" in md
        assert "rising" in md

    def test_c1_shows_direction_modifier(self):
        analysis = _base_analysis()
        analysis["components"]["breadth_level_trend"]["ma8_direction"] = "falling"
        analysis["components"]["breadth_level_trend"]["direction_modifier"] = -10
        md = _generate_to_string(analysis)
        assert "Direction Modifier" in md
        assert "-10" in md

    def test_c1_omits_modifier_when_zero(self):
        analysis = _base_analysis()
        analysis["components"]["breadth_level_trend"]["ma8_direction"] = "rising"
        analysis["components"]["breadth_level_trend"]["direction_modifier"] = 0
        md = _generate_to_string(analysis)
        # Extract C1 section (between "### 1." and "### 2.")
        c1_start = md.index("### 1.")
        c1_end = md.index("### 2.")
        c1_section = md[c1_start:c1_end]
        assert "Direction Modifier" not in c1_section


# =========================================================================
# Step 3 tests: multi-window divergence in C6
# =========================================================================


class TestStep3MultiWindow:
    def test_c6_shows_both_windows(self):
        analysis = _base_analysis()
        analysis["components"]["divergence"]["window_60d"] = {
            "score": 70,
            "divergence_type": "Healthy alignment (both rising)",
            "lookback_days": 60,
        }
        analysis["components"]["divergence"]["window_20d"] = {
            "score": 30,
            "divergence_type": "Consistent decline (both falling)",
            "lookback_days": 20,
        }
        md = _generate_to_string(analysis)
        assert "60-Day Window" in md
        assert "20-Day Window" in md

    def test_c6_shows_early_warning(self):
        analysis = _base_analysis()
        analysis["components"]["divergence"]["window_60d"] = {
            "score": 70,
            "divergence_type": "Healthy",
            "lookback_days": 60,
        }
        analysis["components"]["divergence"]["window_20d"] = {
            "score": 20,
            "divergence_type": "Bearish",
            "lookback_days": 20,
        }
        analysis["components"]["divergence"]["early_warning"] = True
        md = _generate_to_string(analysis)
        assert "Early Warning" in md

    def test_c6_fallback_no_windows(self):
        """Old format without window keys should still render."""
        md = _generate_to_string(_base_analysis())
        assert "S&P 500 vs Breadth Divergence" in md


# =========================================================================
# Step 4 tests: Score Trend section
# =========================================================================


class TestStep4ScoreTrend:
    def test_trend_shown_when_entries_ge_2(self):
        analysis = _base_analysis()
        analysis["trend_summary"] = {
            "entries": [
                {"data_date": "2025-01-13", "composite_score": 55.0},
                {"data_date": "2025-01-14", "composite_score": 60.0},
            ],
            "delta": 5.0,
            "direction": "improving",
        }
        md = _generate_to_string(analysis)
        assert "## Score Trend" in md
        assert "improving" in md.lower() or "Improving" in md

    def test_trend_shown_as_na_when_single_entry(self):
        analysis = _base_analysis()
        analysis["trend_summary"] = {
            "entries": [{"data_date": "2025-01-14", "composite_score": 60.0}],
            "delta": 0,
            "direction": "stable",
        }
        md = _generate_to_string(analysis)
        assert "## Score Trend" in md
        assert "N/A" in md

    def test_trend_omitted_when_zero_entries(self):
        analysis = _base_analysis()
        analysis["trend_summary"] = {
            "entries": [],
            "delta": 0,
            "direction": "stable",
        }
        md = _generate_to_string(analysis)
        assert "## Score Trend" not in md

    def test_trend_omitted_when_missing(self):
        md = _generate_to_string(_base_analysis())
        assert "## Score Trend" not in md


# =========================================================================
# Improvement 2 tests: Guidance caution when direction modifiers negative
# =========================================================================


class TestGuidanceCaution:
    def test_guidance_shows_caution_when_modifiers_negative(self):
        analysis = _base_analysis()
        analysis["components"]["breadth_level_trend"]["direction_modifier"] = -10
        analysis["components"]["ma_crossover"]["direction_modifier"] = -10
        md = _generate_to_string(analysis)
        assert "Caution" in md

    def test_guidance_no_caution_when_modifiers_zero(self):
        analysis = _base_analysis()
        analysis["components"]["breadth_level_trend"]["direction_modifier"] = 0
        analysis["components"]["ma_crossover"]["direction_modifier"] = 0
        md = _generate_to_string(analysis)
        assert "Caution" not in md

    def test_guidance_caution_when_only_c1_negative(self):
        analysis = _base_analysis()
        analysis["components"]["breadth_level_trend"]["direction_modifier"] = -10
        analysis["components"]["ma_crossover"]["direction_modifier"] = 0
        md = _generate_to_string(analysis)
        assert "Caution" in md

    def test_guidance_caution_when_only_c2_negative(self):
        analysis = _base_analysis()
        analysis["components"]["breadth_level_trend"]["direction_modifier"] = 0
        analysis["components"]["ma_crossover"]["direction_modifier"] = -10
        md = _generate_to_string(analysis)
        assert "Caution" in md


# =========================================================================
# Condition-dependent actions when Caution fires
# =========================================================================


class TestConditionDependentActions:
    def test_cautionary_actions_injected_when_falling(self):
        analysis = _base_analysis()
        analysis["components"]["breadth_level_trend"]["direction_modifier"] = -10
        analysis["components"]["ma_crossover"]["direction_modifier"] = -10
        md = _generate_to_string(analysis)
        assert "Reduce new position sizes" in md
        assert "Tighten stop-loss levels" in md

    def test_no_cautionary_actions_when_healthy(self):
        analysis = _base_analysis()
        analysis["components"]["breadth_level_trend"]["direction_modifier"] = 0
        analysis["components"]["ma_crossover"]["direction_modifier"] = 0
        md = _generate_to_string(analysis)
        assert "Reduce new position sizes" not in md
        assert "Tighten stop-loss levels" not in md

    def test_cautionary_actions_when_only_c1_negative(self):
        analysis = _base_analysis()
        analysis["components"]["breadth_level_trend"]["direction_modifier"] = -10
        analysis["components"]["ma_crossover"]["direction_modifier"] = 0
        md = _generate_to_string(analysis)
        assert "Reduce new position sizes" in md


# =========================================================================
# C6 score displays as int (not float)
# =========================================================================


class TestC6ScoreIntDisplay:
    def test_c6_float_score_renders_as_int(self):
        analysis = _base_analysis()
        analysis["composite"]["component_scores"]["divergence"]["score"] = 62.0
        md = _generate_to_string(analysis)
        # Should show "62" not "62.0" in the score column
        assert " 62.0 " not in md
        assert " 62 " in md

    def test_integer_scores_unchanged(self):
        analysis = _base_analysis()
        md = _generate_to_string(analysis)
        # C1 score=80 should still appear as 80
        assert " 80 " in md

"""Tests for the composite scoring engine."""

from scorer import COMPONENT_WEIGHTS, calculate_composite_score


def _all_scores(value=50):
    """Return component_scores dict with all components set to *value*."""
    return {k: value for k in COMPONENT_WEIGHTS}


def _all_available():
    return {k: True for k in COMPONENT_WEIGHTS}


def _all_unavailable():
    return {k: False for k in COMPONENT_WEIGHTS}


class TestWeightRedistribution:
    """When a component is data_available=False, its weight is redistributed."""

    def test_all_available_uses_original_weights(self):
        scores = _all_scores(60)
        result = calculate_composite_score(scores, _all_available())
        assert result["composite_score"] == 60.0

    def test_one_excluded_redistributes_weight(self):
        """Excluding cycle_position (weight 0.20) redistributes to remaining."""
        scores = _all_scores(60)
        avail = _all_available()
        avail["cycle_position"] = False

        result = calculate_composite_score(scores, avail)
        # All remaining scores are 60, so composite should still be 60
        assert result["composite_score"] == 60.0

    def test_excluded_component_different_score(self):
        """Excluding a low-scoring component raises the composite."""
        scores = _all_scores(80)
        scores["cycle_position"] = 20  # Low score
        avail = _all_available()
        avail["cycle_position"] = False

        result = calculate_composite_score(scores, avail)
        # cycle_position excluded, so 20 is ignored; rest are 80
        assert result["composite_score"] == 80.0

    def test_effective_weights_sum_to_one(self):
        scores = _all_scores(50)
        avail = _all_available()
        avail["divergence"] = False
        avail["historical_percentile"] = False

        result = calculate_composite_score(scores, avail)
        total_eff = sum(
            c["effective_weight"]
            for c in result["component_scores"].values()
            if c.get("data_available", True)
        )
        assert abs(total_eff - 1.0) < 1e-9

    def test_excluded_components_listed(self):
        scores = _all_scores(50)
        avail = _all_available()
        avail["cycle_position"] = False

        result = calculate_composite_score(scores, avail)
        assert "Peak/Trough Cycle Position" in result["excluded_components"]

    def test_all_unavailable_returns_neutral(self):
        """All components unavailable -> score=50, zone=Neutral, warning."""
        scores = _all_scores(70)
        result = calculate_composite_score(scores, _all_unavailable())
        assert result["composite_score"] == 50.0
        assert result["zone"] == "Neutral"
        assert len(result["excluded_components"]) == 6

    def test_strongest_weakest_exclude_unavailable(self):
        """strongest/weakest should only consider available components."""
        scores = {
            "breadth_level_trend": 90,
            "ma_crossover": 30,
            "cycle_position": 10,  # lowest but unavailable
            "bearish_signal": 70,
            "historical_percentile": 50,
            "divergence": 80,
        }
        avail = _all_available()
        avail["cycle_position"] = False

        result = calculate_composite_score(scores, avail)
        assert result["strongest_health"]["component"] == "breadth_level_trend"
        assert result["weakest_health"]["component"] == "ma_crossover"

    def test_backward_compat_no_data_availability(self):
        """When data_availability is None, all components treated as available."""
        scores = _all_scores(60)
        result = calculate_composite_score(scores)
        assert result["composite_score"] == 60.0
        assert len(result.get("excluded_components", [])) == 0

    def test_component_scores_have_effective_weight(self):
        scores = _all_scores(50)
        result = calculate_composite_score(scores, _all_available())
        for _key, comp in result["component_scores"].items():
            assert "effective_weight" in comp
            assert "data_available" in comp

    def test_excluded_component_has_zero_effective_weight(self):
        scores = _all_scores(50)
        avail = _all_available()
        avail["divergence"] = False
        result = calculate_composite_score(scores, avail)
        assert result["component_scores"]["divergence"]["effective_weight"] == 0.0

"""Tests for Component 1: Trend Level Calculator — 8MA direction modifier."""

from calculators.trend_level_calculator import calculate_breadth_level_trend


def _make_direction_rows(make_row, n, ma8_5d_ago, ma8_latest, **kwargs):
    """Create n rows where rows[-6] and rows[-1] have specified 8MA values."""
    rows = [make_row(Breadth_Index_8MA=ma8_latest, **kwargs) for _ in range(n)]
    rows[0]["Breadth_Index_8MA"] = ma8_5d_ago
    return rows


class TestDirectionModifier:
    """8MA direction modifier tests."""

    def test_falling_high_level_penalty(self, make_row):
        """8MA falling from high level (>0.60) -> -10."""
        rows = _make_direction_rows(make_row, 6, ma8_5d_ago=0.72, ma8_latest=0.65)
        result = calculate_breadth_level_trend(rows)
        assert result["ma8_direction"] == "falling"
        assert result["direction_modifier"] == -10

    def test_falling_low_level_bonus(self, make_row):
        """8MA falling at low level (<0.40) -> +5."""
        rows = _make_direction_rows(make_row, 6, ma8_5d_ago=0.38, ma8_latest=0.35)
        result = calculate_breadth_level_trend(rows)
        assert result["ma8_direction"] == "falling"
        assert result["direction_modifier"] == 5

    def test_rising_low_level_bonus(self, make_row):
        """8MA rising at level < 0.60 -> +5."""
        rows = _make_direction_rows(make_row, 6, ma8_5d_ago=0.40, ma8_latest=0.50)
        result = calculate_breadth_level_trend(rows)
        assert result["ma8_direction"] == "rising"
        assert result["direction_modifier"] == 5

    def test_rising_high_level_no_modifier(self, make_row):
        """8MA rising at level >= 0.60 -> 0 (no modifier)."""
        rows = _make_direction_rows(make_row, 6, ma8_5d_ago=0.62, ma8_latest=0.68)
        result = calculate_breadth_level_trend(rows)
        assert result["ma8_direction"] == "rising"
        assert result["direction_modifier"] == 0

    def test_insufficient_rows_no_modifier(self, make_row):
        """Fewer than 6 rows -> no direction modifier."""
        rows = [make_row(Breadth_Index_8MA=0.55) for _ in range(3)]
        result = calculate_breadth_level_trend(rows)
        assert result.get("direction_modifier", 0) == 0

    def test_flat_no_modifier(self, make_row):
        """8MA unchanged -> no modifier."""
        rows = [make_row(Breadth_Index_8MA=0.55) for _ in range(6)]
        result = calculate_breadth_level_trend(rows)
        assert result.get("direction_modifier", 0) == 0

    def test_score_clamped_0_100(self, make_row):
        """Score should remain in 0-100 after modifier."""
        rows = _make_direction_rows(
            make_row,
            6,
            ma8_5d_ago=0.10,
            ma8_latest=0.15,
            Breadth_200MA_Trend=-1,
        )
        result = calculate_breadth_level_trend(rows)
        assert 0 <= result["score"] <= 100

    def test_modifier_affects_final_score(self, make_row):
        """Direction modifier should change the final score."""
        # Flat -> no modifier
        flat_rows = [make_row(Breadth_Index_8MA=0.65, Breadth_200MA_Trend=1) for _ in range(6)]
        result_flat = calculate_breadth_level_trend(flat_rows)

        # Falling from high -> -10
        falling_rows = _make_direction_rows(
            make_row,
            6,
            ma8_5d_ago=0.70,
            ma8_latest=0.65,
            Breadth_200MA_Trend=1,
        )
        result_falling = calculate_breadth_level_trend(falling_rows)

        assert result_falling["score"] < result_flat["score"]


class TestDirectionModifierBoundary:
    """Boundary tests for 0.40 and 0.60 thresholds (strict inequality)."""

    def test_falling_at_exactly_060_no_modifier(self, make_row):
        """8MA=0.60 falling: > 0.60 is False -> modifier 0."""
        rows = _make_direction_rows(make_row, 6, ma8_5d_ago=0.62, ma8_latest=0.60)
        result = calculate_breadth_level_trend(rows)
        assert result["ma8_direction"] == "falling"
        assert result["direction_modifier"] == 0

    def test_falling_at_exactly_040_no_modifier(self, make_row):
        """8MA=0.40 falling: < 0.40 is False -> modifier 0."""
        rows = _make_direction_rows(make_row, 6, ma8_5d_ago=0.42, ma8_latest=0.40)
        result = calculate_breadth_level_trend(rows)
        assert result["ma8_direction"] == "falling"
        assert result["direction_modifier"] == 0

    def test_rising_at_exactly_060_no_modifier(self, make_row):
        """8MA=0.60 rising: < 0.60 is False -> modifier 0."""
        rows = _make_direction_rows(make_row, 6, ma8_5d_ago=0.58, ma8_latest=0.60)
        result = calculate_breadth_level_trend(rows)
        assert result["ma8_direction"] == "rising"
        assert result["direction_modifier"] == 0

---
name: market-breadth-analyzer
description: Quantifies market breadth health using TraderMonty's public CSV data. Generates a 0-100 composite score across 6 components (100 = healthy). No API key required. Use when user asks about market breadth, participation rate, advance-decline health, whether the rally is broad-based, or general market health assessment.
---

# Market Breadth Analyzer Skill

## Purpose

Quantify market breadth health using a data-driven 6-component scoring system (0-100). Uses TraderMonty's publicly available CSV data to measure how broadly the market is participating in a rally or decline.

**Score direction:** 100 = Maximum health (broad participation), 0 = Critical weakness.

**No API key required** - uses freely available CSV data from GitHub Pages.

## When to Use This Skill

**English:**
- User asks "Is the market rally broad-based?" or "How healthy is market breadth?"
- User wants to assess market participation rate
- User asks about advance-decline indicators or breadth thrust
- User wants to know if the market is narrowing (fewer stocks participating)
- User asks about equity exposure levels based on breadth conditions

**Japanese:**
- 「マーケットブレッドスはどうですか？」「市場の参加率は？」
- 「上昇は広がっている？」「一部の銘柄だけの上昇？」
- ブレッドス指標に基づくエクスポージャー判断
- 市場の健康度をデータで確認したい

## Prerequisites

- **Python 3.9+** with `requests` library (for fetching CSV data)
- **Internet access** to reach GitHub Pages URLs
- **No API keys required** - uses freely available public CSV data

## Difference from Breadth Chart Analyst

| Aspect | Market Breadth Analyzer | Breadth Chart Analyst |
|--------|------------------------|----------------------|
| Data Source | CSV (automated) | Chart images (manual) |
| API Required | None | None |
| Output | Quantitative 0-100 score | Qualitative chart analysis |
| Components | 6 scored dimensions | Visual pattern recognition |
| Repeatability | Fully reproducible | Analyst-dependent |

---

## Execution Workflow

### Phase 1: Execute Python Script

Run the analysis script. If using a nested or date-stamped `--output-dir` in cron runs, create it first; the history writer expects the directory to already exist.

```bash
mkdir -p reports/<routine-or-date>
python3 skills/market-breadth-analyzer/scripts/market_breadth_analyzer.py \
  --detail-url "https://tradermonty.github.io/market-breadth-analysis/market_breadth_data.csv" \
  --summary-url "https://tradermonty.github.io/market-breadth-analysis/market_breadth_summary.csv" \
  --output-dir reports/<routine-or-date>
```

For a simple ad-hoc run, omit `--output-dir` or use an existing directory. In scheduled cron runs from the repository root, prefer a repo-relative output directory such as `reports/after-close-YYYY-MM-DD` rather than an absolute path. If an absolute nested `--output-dir` unexpectedly fails at the history-writing step despite the directory existing, rerun once with the equivalent repo-relative path before treating the breadth analysis as unavailable.

The script will:
1. Fetch detail CSV (~2,500 rows, 2016-present) and summary CSV (8 metrics)
2. Validate data freshness (warn if > 5 days old)
3. Calculate all 6 component scores (with automatic weight redistribution if any component lacks data)
4. Generate composite score with zone classification
5. Track score history and compute trend (improving/deteriorating/stable)
6. Output JSON and Markdown reports

### Phase 2: Present Results

Present the generated Markdown report to the user, highlighting:
- Composite score and health zone
- Strongest and weakest components
- Recommended equity exposure level
- Key breadth levels to watch
- Any data freshness warnings

---

## 6-Component Scoring System

| # | Component | Weight | Key Signal |
|---|-----------|--------|------------|
| 1 | Breadth Level & Trend | **25%** | Current 8MA level + 200MA trend direction + 8MA direction modifier |
| 2 | 8MA vs 200MA Crossover | **20%** | Momentum via MA gap and direction |
| 3 | Peak/Trough Cycle | **20%** | Position in breadth cycle |
| 4 | Bearish Signal | **15%** | Backtested bearish signal flag |
| 5 | Historical Percentile | **10%** | Current vs full history distribution |
| 6 | S&P 500 Divergence | **10%** | Multi-window (20d + 60d) price vs breadth divergence |

**Weight Redistribution:** If any component lacks sufficient data (e.g., no peak/trough markers detected), it is excluded and its weight is proportionally redistributed among the remaining components. The report shows both original and effective weights.

**Score History:** Composite scores are persisted across runs (keyed by data date). The report includes a trend summary (improving/deteriorating/stable) when multiple observations are available.

## Health Zone Mapping (100 = Healthy)

| Score | Zone | Equity Exposure | Action |
|-------|------|-----------------|--------|
| 80-100 | Strong | 90-100% | Full position, growth/momentum favored |
| 60-79 | Healthy | 75-90% | Normal operations |
| 40-59 | Neutral | 60-75% | Selective positioning, tighten stops |
| 20-39 | Weakening | 40-60% | Profit-taking, raise cash |
| 0-19 | Critical | 25-40% | Capital preservation, watch for trough |

---

## Data Sources

**Detail CSV:** `market_breadth_data.csv`
- ~2,500 rows from 2016-02 to present
- Columns: Date, S&P500_Price, Breadth_Index_Raw, Breadth_Index_200MA, Breadth_Index_8MA, Breadth_200MA_Trend, Bearish_Signal, Is_Peak, Is_Trough, Is_Trough_8MA_Below_04

**Summary CSV:** `market_breadth_summary.csv`
- 8 aggregate metrics (average peaks, average troughs, counts, analysis period)

Both are publicly hosted on GitHub Pages - no authentication required.

## Output Files

- JSON: `market_breadth_YYYY-MM-DD_HHMMSS.json`
- Markdown: `market_breadth_YYYY-MM-DD_HHMMSS.md`
- History: `market_breadth_history.json` (persists across runs, max 20 entries)

## Reference Documents

### `references/breadth_analysis_methodology.md`
- Full methodology with component scoring details
- Threshold explanations and zone definitions
- Historical context and interpretation guide

### When to Load References
- **First use:** Load methodology reference for framework understanding
- **Regular execution:** References not needed - script handles scoring

## Same-run provenance

Pass `--run-id daily-20261017-unique` to both required market-regime-daily
producers. Without it each producer generates its own unique ID. JSON includes
schema 1.0 `data_provenance` with actual primary timeseries business date,
UTC retrieval time, source and run ID. Unknown adjustment, point-in-time,
survivorship and corporate-action flags are false. Auxiliary summary dates
are unverified; this does not certify the freshness of every composite input.
Missing or malformed primary dates stop report generation.

Pass the exact JSON paths printed by these invocations to Exposure Coach's
`--require-provenance --run-id … --as-of …` mode. Run IDs are caller-provided
consistency markers, not authenticity or point-in-time proof.