kanchi-dividend-sop

어떤 스킬인가요?

Convert Kanchi-style dividend investing into a repeatable US-stock operating procedure. Use when users ask for かんち式配当投資, dividend screening, dividend growth quality checks, PERxPBR adaptation for US sectors, pullback limit-order planning, or one-page stock memo creation. Covers screening, deep dive, entry planning, and post-purchase monitoring cadence.

설치 방법

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

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

스킬 원문 파일

interface:
  display_name: "Kanchi Dividend Sop"
  short_description: "Help with Kanchi Dividend Sop tasks"

# Default Thresholds

Use these defaults when users do not provide custom risk settings.

## Baseline (US Stock-Focused)

| Category | Metric | Default | Interpretation |
|---|---|---:|---|
| Yield | Forward dividend yield | >= 3.5% | Kanchi Step 1 core filter |
| Yield trap | Extreme yield flag | >= 8.0% | Force deep dive before pass |
| Growth | Revenue CAGR (5y) | > 0% | Basic business expansion check |
| Growth | EPS CAGR (5y) | > 0% | Earnings trend health |
| Dividend trend | Dividend growth (5y) | Non-declining | Allow flat only with strong safety |
| Safety | EPS payout ratio | <= 70% | `70-85%` = caution |
| Safety | FCF payout ratio | <= 80% | `>100%` = high risk |
| Balance sheet | Net debt trend | Not persistently rising | Rising 3 periods = warning |
| Balance sheet | Interest coverage | >= 3.0x | `<2.5x` = caution |
| Entry trigger | Yield alpha vs 5y average | +0.5pp | Default Kanchi Step 5 pullback threshold |

## Instrument-Specific Notes

Use these denominator replacements for coverage checks:

| Instrument type | Primary denominator | Notes |
|---|---|---|
| Stock | FCF | Use `CFO - CapEx` |
| REIT | FFO/AFFO | Prefer AFFO if available |
| BDC | NII | Compare NII with distribution |
| ETF | Fund-level distribution coverage unavailable in many cases | Focus on methodology and holdings quality |

## Objective Tuning

Apply profile-specific adjustments:

| Profile | Yield floor | Safety bias | Notes |
|---|---:|---|---|
| Income now | 4.0% | Tight safety checks | Avoid overconcentration in one high-yield sector |
| Balanced | 3.0-3.5% | Medium | Blend current income and dividend growth |
| Growth first | 1.5-2.5% | High quality first | Accept lower initial yield for higher dividend CAGR |

## Step 5 Alpha Tuning

Use this range for yield-trigger alpha:
- Stable mega-cap compounders: `+0.3pp`.
- Default baseline: `+0.5pp`.
- Higher-volatility names: `+0.8pp` to `+1.0pp`.

## Entry Signal Interpretation

When using `build_entry_signals.py`, interpret signals as follows:

| Signal | Meaning | Action |
|---|---|---|
| TRIGGERED | Current price <= buy target price | Ready for first tranche (40%) if thesis intact |
| WAIT | Current price > buy target price | Monitor; set limit order at buy target |
| ASSUMPTION-REQUIRED | Missing data (yield history or dividend) | Manual research needed before entry |

### Drop Needed Percentage

- `0%`: Price already at or below target.
- `1-10%`: Near entry zone; consider scaling in if pullback accelerates.
- `>10%`: Significant gap; wait for pullback or reassess target.

## Portfolio Constraints

Use these as practical defaults:
- Max positions: `15-30`.
- Max single position at cost: `<= 8%`.
- Max sector exposure: `<= 25%`.
- High-yield bucket (REIT/telecom/utilities combined): `<= 35%`.

Override only when user explicitly chooses a different policy.

# Sector-Specific Step 2 Modules (WS-4)

Step 2 (growth & safety) is sector-dispatched. This file holds the
**Step 2 safety indicators by sector**. It is the companion to
`valuation-and-one-off-checks.md`, which owns **Step 3 valuation** and
**Step 4 one-off** logic. Responsibility boundary (MN-2):

| Concern | Owner file |
|---|---|
| Step 2 safety indicators by sector | `sector-step2-modules.md` (this file) |
| Step 3 valuation mapping | `valuation-and-one-off-checks.md` |
| Step 4 backward one-off checklist | `valuation-and-one-off-checks.md` |
| Step 4b forward structural-event scan | `event_scanner.py` + SKILL.md Step 4b |

The deterministic thresholds are the SSOT in `scripts/thresholds.py`; the
numbers quoted below are human-readable mirrors. The script
`scripts/payout_safety.py` implements the dispatch.

## Why a uniform triad fails

FCF payout is the right anchor for consumer/industrial cash-cows, but:

- **Banks**: FCF is not a meaningful concept; capital adequacy and credit
  quality govern dividend safety.
- **Regulated utilities**: FCF is *structurally negative* (rate-base capex
  funded by debt/equity, recovered through rates). Auto-FAIL on negative
  FCF wrongly rejects healthy regulated names — this was the income-now vs
  balanced inconsistency in the 2026-05 runs.
- **Insurers**: GAAP earnings are noisy; statutory capital, reserve
  development and the combined ratio govern dividend capacity.

## Consumer / Industrial / Communication / default

Primary: **Adjusted-EPS payout + FCF payout**.

- Adjusted-EPS payout ≤ 70% PASS; 70–85% CAUTION; > 85% FAIL.
- FCF payout ≤ 80% PASS; > 100% FAIL.
- GAAP↔Adjusted EPS divergence > 25% ⇒ Step-4 one-off flag (MKC de Mexico
  non-cash remeasurement gain is the golden case).
- `adjusted_eps_source = UNAVAILABLE` ⇒ cap HOLD-REVIEW (never silent PASS).

## Banks

Primary: **EPS payout + capital + credit trend** (FCF ignored).

| Indicator | Caution / blocker condition |
|---|---|
| CET1 ratio | missing ⇒ `bank_capital_unavailable`; low vs peers ⇒ CAUTION |
| NPL trend | deteriorating ⇒ `bank_npl_nco_deteriorating` (CAUTION) |
| NCO trend | deteriorating ⇒ same blocker |
| Criticized / classified loans | rising ⇒ CAUTION |
| CRE / construction concentration | high or unavailable ⇒ `bank_cre_concentration_unavailable` (CAUTION) |
| Deposit cost / beta | rising sharply ⇒ CAUTION |
| Uninsured deposit % / AOCI hit | elevated ⇒ CAUTION |

Golden case: **OZK** — strong serial raiser, low payout, but NPL 0.20%→0.90%
and NCO 0.25%→0.57% (Q1-2026) ⇒ **PASS-CAUTION**, not clean PASS.

## Regulated Utilities

Primary: **EPS payout + FFO/debt + allowed ROE + rate-case + equity issuance**.
Negative FCF is **not** an auto-FAIL.

| Indicator | Caution / blocker condition |
|---|---|
| FFO / debt | missing ⇒ `utility_ffo_debt_unavailable` (CAUTION); weak ⇒ CAUTION |
| Allowed ROE | falling / adverse order ⇒ CAUTION |
| Rate-case status | adverse / pending-adverse ⇒ `utility_rate_case_adverse` (HOLD-REVIEW) |
| Equity issuance risk | high / dilutive ⇒ `utility_equity_issuance_risk` (CAUTION) |
| EPS payout | > 100% ⇒ FAIL |

Golden cases: **WTRG** (AWK all-stock merger — Step 4b event, not a clean
utility PASS), **HTO** (Quadvest + ~$2.7B capex funding ⇒ equity-issuance
risk, low-priority).

## Insurers

Primary: **Operating-EPS payout + combined ratio + reserve development +
statutory capital**. Separate regular vs special dividends (ORI pays large
specials; the headline ~9% yield is a special-inclusive artifact — Step 1
must use the regular forward yield, ≈3.2%).

| Indicator | Caution / blocker condition |
|---|---|
| Combined ratio | missing ⇒ `insurer_combined_ratio_unavailable`; > 100 ⇒ CAUTION |
| Reserve development | adverse ⇒ `insurer_reserve_development_adverse` (HOLD-REVIEW) |
| Statutory capital | weak ⇒ HOLD-REVIEW |
| Special vs regular split | specials must be excluded from the income base |

Golden case: **ORI** — regular yield clears the 3% (balanced) floor →
Step-1 PASS → insurer-module HOLD-REVIEW (lumpy specials), NOT "trap /
below floor".

# One-Page Stock Memo Template

Use one memo per ticker. Keep each section concise and evidence-based.

```markdown
# [Ticker] - Dividend Memo

## Business Model
- Revenue engine:
- Core customer segments:
- Structural moat/risk:

## Dividend Policy
- Current forward yield:
- Dividend growth track record:
- Historical cuts/suspensions:

## Dividend Safety
- EPS payout ratio:
- FCF payout ratio (or FFO/NII coverage):
- Net debt trend:
- Interest coverage:
- Verdict: PASS / CAUTION / FAIL

## Kanchi Step 3 Valuation
- Method used: (PERxPBR / P-FFO / P-FCF / forward PE)
- Current value:
- Historical reference:
- Valuation verdict:

## Kanchi Step 4 One-Off Check
- Key finding (one sentence):
- Verdict: PASS / FAIL

## Verdict (WS-5 actionable tier)
- Tier: CLEAN-PASS / PASS-CAUTION / CONDITIONAL-PASS / HOLD-REVIEW / STEP1-RECHECK / FAIL
- Verdict reasons:

## Pre-Order Blockers (WS-6)
- pre_order_blockers[]: <list, or "none">
- t1_blocked: true / false
- If any blocker is unresolved OR t1_blocked: **T1 is blocked or downsized
  to a ≤20% tracking tranche** (never the full 40%). Resolve / acknowledge
  each blocker (status: blocked → acknowledged → cleared) before sizing up.

## Entry Plan (Kanchi Step 5)
- Trigger type: Yield / Valuation / Both
  - **Yield trigger**: current yield >= 5y avg yield + alpha (default +0.5pp)
  - **Valuation trigger**: price reaches target multiple (P/E, P/FFO, P/FCF)
- Buy zone: $[lower] - $[upper]
- Split orders: 40% / 30% / 30%  (T1 gated by Pre-Order Blockers above)
- Invalidation condition:

## Maximum Risk
- "If this happens, thesis is broken":

## Provenance (audit trail)
```yaml
provenance:
  price_source:
  price_asof:
  dividend_source:        # fmp_stock_dividend | issuer_ir
  dividend_dates_used: []
  payout_source:          # FMP | issuer non-GAAP recon | UNAVAILABLE
  event_scan_result:      # CLEAN_CONFIRMED | NO_EVENT_FOUND | MAJOR_EVENT | ...
  event_scan_checked_at:
  unresolved_blockers: []
  evidence_refs:
    - claim:
      source_type:        # issuer_ir | sec_filing | exchange | wire | portal
      source_url:
      checked_at:
      raw_value:
      normalized_value:
      confidence:          # high | medium | low
```

## Run Context (profile guard — never reuse a 3%-run result in a 4%-run)
```yaml
run_context:
  profile:                # income-now | balanced | growth-first
  yield_floor_pct:
  safety_bias:            # tight | medium
  universe_source:
  excluded_asset_types: []
```
```

Do not skip the invalidation condition, the provenance block, or the
pre-order blocker gate.

# Valuation And One-Off Checks

Use this file for Kanchi Step 3 and Step 4 decisions.

## Step 3: Valuation Mapping By Sector

| Sector/Type | Primary metric | Secondary metric | Pass condition example |
|---|---|---|---|
| Banks/Insurers | `P/TBV` | `PER x PBR` (compatibility) + historical percentile | Reject when clearly above historical fair band |
| REIT | `P/FFO` or `P/AFFO` | Implied cap rate vs peers | Prefer below own 5y median multiple |
| Asset-light growth | Forward `P/E` | `P/FCF` and 5y range | Require at least one valuation metric near lower half of range |
| Mature cash cows | `P/FCF` | Dividend yield vs 5y average | Prefer yield above historical mean with intact fundamentals |

Note:
- Kanchi's original Japan-style filter uses `PER x PBR`.
- For US banks, `P/TBV` is typically more robust because tangible book is the standard anchor.

If metrics disagree, choose `HOLD-FOR-REVIEW` instead of forcing `PASS`.

## Step 4: One-Off Profit Checklist

Mark each item `YES/NO`. Two or more `YES` values require downgrade or reject.

1. Profit jump driven by asset sales or disposal gains.
2. EPS lifted by legal settlement or tax one-time effects.
3. Margin expansion without matching revenue quality.
4. Management repeatedly labels key profit items as "non-recurring."
5. Dividend support appears linked to debt increase or asset liquidation.

## Step 4 Output Format

Use this exact compact output in reports:

```text
Step4 verdict: FAIL
Reason: EPS uplift mostly from one-time asset sale; recurring margin trend still weak.
```

Keep each reason to one sentence.

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

#!/usr/bin/env python3
"""Build Kanchi Step 5 entry signals using live FMP data."""

from __future__ import annotations

import argparse
import csv
import json
import os
from datetime import date, datetime, timezone
from pathlib import Path
from typing import Any

from dividend_basis import analyze_dividends, step1_decision
from event_scanner import ScanResult, apply_event_cap
from fmp_client import FMPClient as SharedFMPClient
from payout_safety import assess_payout_safety
from thresholds import SCHEMA_VERSION, VERDICTS
from verdict import build_run_context, synthesize_verdict


def parse_ticker_csv(raw: str) -> list[str]:
    tickers: list[str] = []
    for part in raw.split(","):
        value = part.strip().upper()
        if not value:
            continue
        if value not in tickers:
            tickers.append(value)
    return tickers


def load_tickers(input_path: Path | None, tickers_csv: str | None) -> list[str]:
    if tickers_csv:
        return parse_ticker_csv(tickers_csv)

    if not input_path:
        return []

    payload = json.loads(input_path.read_text())
    tickers: list[str] = []

    def append_ticker(value: Any) -> None:
        ticker = str(value or "").strip().upper()
        if ticker and ticker not in tickers:
            tickers.append(ticker)

    raw_candidates = payload.get("candidates")
    if isinstance(raw_candidates, list):
        for item in raw_candidates:
            if isinstance(item, dict):
                append_ticker(item.get("ticker") or item.get("symbol"))
            else:
                append_ticker(item)

    # value-dividend-screener and dividend-growth-pullback-screener both
    # publish their ranked JSON rows as ``stocks[].symbol``. Accept that
    # contract directly so workflow handoff does not depend on a prose-only
    # manual rename to candidates/ticker.
    raw_stocks = payload.get("stocks")
    if isinstance(raw_stocks, list):
        for item in raw_stocks:
            if isinstance(item, dict):
                append_ticker(item.get("symbol") or item.get("ticker"))
            else:
                append_ticker(item)

    raw_tickers = payload.get("tickers")
    if isinstance(raw_tickers, list):
        for item in raw_tickers:
            append_ticker(item)

    return tickers


def to_float(value: Any) -> float | None:
    if value is None:
        return None
    if isinstance(value, bool):
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def normalize_metrics_yields(metrics: list[dict[str, Any]], max_points: int = 5) -> list[float]:
    yields_pct: list[float] = []
    for item in metrics:
        raw = to_float(item.get("dividendYield"))
        if raw is None or raw <= 0:
            continue
        # FMP usually returns decimal (0.035). Guard for percent-style values.
        normalized = raw * 100 if raw <= 1.5 else raw
        yields_pct.append(normalized)
        if len(yields_pct) >= max_points:
            break
    return yields_pct


def average(values: list[float]) -> float | None:
    if not values:
        return None
    return sum(values) / len(values)


class FMPClient(SharedFMPClient):
    """Kanchi domain methods backed by the generated shared transport."""

    def __init__(self, api_key: str, sleep_seconds: float = 0.15, timeout: int = 30):
        super().__init__(api_key=api_key)
        self.sleep_seconds = sleep_seconds
        self.RATE_LIMIT_DELAY = sleep_seconds
        self.timeout = timeout

    @property
    def api_calls(self) -> int:
        return self.api_calls_made

    def get_batch_quotes(self, tickers: list[str]) -> dict[str, dict[str, Any]]:
        """Fetch quotes for all tickers (one /stable quote request per symbol).

        Comma-batched /stable quote requests silently return [], so each symbol
        is requested individually.
        """
        result: dict[str, dict[str, Any]] = {}
        for ticker in tickers:
            data = self._get("quote", {"symbol": ticker})
            if not isinstance(data, list) or not data:
                continue
            row = data[0]
            if not isinstance(row, dict):
                continue
            symbol = str(row.get("symbol", "")).strip().upper()
            if symbol:
                result[symbol] = row
        return result

    def get_batch_profiles(self, tickers: list[str]) -> dict[str, dict[str, Any]]:
        """Fetch profiles for all tickers (one /stable profile request per symbol).

        /stable renamed the trailing dividend field lastDiv -> lastDividend; a
        lastDiv alias is restored so downstream consumers are unchanged.
        """
        result: dict[str, dict[str, Any]] = {}
        for ticker in tickers:
            data = self._get("profile", {"symbol": ticker})
            if not isinstance(data, list) or not data:
                continue
            row = data[0]
            if not isinstance(row, dict):
                continue
            if "lastDiv" not in row and "lastDividend" in row:
                row["lastDiv"] = row["lastDividend"]
            symbol = str(row.get("symbol", "")).strip().upper()
            if symbol:
                result[symbol] = row
        return result

    def get_ratios(self, ticker: str, limit: int = 10) -> list[dict[str, Any]]:
        """Annual financial ratios (the 5y dividend-yield history series).

        On /stable, dividendYield and dividendPerShare moved off key-metrics
        onto the ratios endpoint (dividendYield is a decimal, e.g. 0.0404).
        """
        data = self._get("ratios", {"symbol": ticker, "limit": limit})
        if isinstance(data, list):
            return [row for row in data if isinstance(row, dict)]
        return []

    def get_stock_dividend(self, ticker: str) -> list[dict[str, Any]]:
        """WS-1: full declared-dividend history (regular + special).

        /stable dividends returns a flat list (the legacy
        historical-price-full/stock_dividend {symbol, historical:[]} wrapper is
        gone). Records still carry date / dividend / declarationDate.
        """
        data = self._get("dividends", {"symbol": ticker})
        if isinstance(data, list):
            return [row for row in data if isinstance(row, dict)]
        return []

    def get_financials(self, ticker: str, sector: str | None = None) -> dict[str, Any]:
        """WS-2 (5th-review F1): minimal financials for the payout triad.

        Adjusted EPS is NOT in FMP -> source UNAVAILABLE by design (the
        consumer path then fail-safes to HOLD-REVIEW; sector paths use
        sector metrics). Sector metric dicts are left None when FMP does
        not expose them, which deterministically raises *_unavailable
        blockers (CAUTION) for manual fill rather than a false PASS.
        """
        inc = self._get("income-statement", {"symbol": ticker, "limit": 1})
        cf = self._get("cash-flow-statement", {"symbol": ticker, "limit": 1})
        gaap_eps = None
        if isinstance(inc, list) and inc:
            # /stable renamed epsdiluted -> epsDiluted; keep both as fallbacks.
            gaap_eps = (
                to_float(inc[0].get("epsDiluted"))
                or to_float(inc[0].get("epsdiluted"))
                or to_float(inc[0].get("eps"))
            )
        fcf_ps = None
        if isinstance(cf, list) and cf:
            fcf = to_float(cf[0].get("freeCashFlow"))
            shares = (
                to_float(inc[0].get("weightedAverageShsOutDil"))
                if (isinstance(inc, list) and inc)
                else None
            )
            if fcf is not None and shares:
                fcf_ps = fcf / shares
        return {
            "sector": sector,
            "gaap_eps": gaap_eps,
            "adjusted_eps": None,
            "adjusted_eps_source": "UNAVAILABLE",
            "fcf_per_share": fcf_ps,
        }


def build_entry_row(
    ticker: str,
    alpha_pp: float,
    quote: dict[str, Any] | None,
    profile: dict[str, Any] | None,
    key_metrics: list[dict[str, Any]],
    dividend_history: list[dict[str, Any]] | None = None,
    floor_pct: float | None = None,
    financials: dict[str, Any] | None = None,
    event_scan: ScanResult | None = None,
    as_of: str | None = None,
) -> dict[str, Any]:
    price = to_float((quote or {}).get("price"))

    # WS-1: prefer the regular run-rate from declared-dividend history over
    # profile.lastDiv (which is a trailing/TTM figure that lagged the latest
    # declared raise -> defect D5 -- and silently bundled specials -> D4).
    basis = None
    if dividend_history:
        basis = analyze_dividends(
            dividend_history,
            price,
            issuer_language=(profile or {}).get("issuer_language"),
            floor_pct=floor_pct,
            as_of_date=as_of,
        )

    annual_dividend = to_float((profile or {}).get("lastDiv"))
    if annual_dividend is None and key_metrics:
        annual_dividend = to_float(key_metrics[0].get("dividendPerShare"))
    if basis is not None and basis.latest_declared_annualized is not None:
        annual_dividend = basis.latest_declared_annualized

    yields_5y = normalize_metrics_yields(key_metrics, max_points=5)
    avg_yield_5y_pct_raw = average(yields_5y)
    avg_yield_5y_pct = round(avg_yield_5y_pct_raw, 2) if avg_yield_5y_pct_raw is not None else None

    target_yield_pct = (
        round(avg_yield_5y_pct + alpha_pp, 2) if avg_yield_5y_pct is not None else None
    )
    buy_target_price = None
    if annual_dividend is not None and target_yield_pct is not None and target_yield_pct > 0:
        buy_target_price = round(annual_dividend / (target_yield_pct / 100), 2)

    current_yield_pct = None
    if annual_dividend is not None and price is not None and price > 0:
        current_yield_pct = round((annual_dividend / price) * 100, 2)

    drop_needed_pct = None
    if price is not None and buy_target_price is not None and price > 0:
        drop_needed_pct = round(max(0.0, (price - buy_target_price) / price * 100), 2)

    signal = "ASSUMPTION-REQUIRED"
    if price is not None and buy_target_price is not None:
        signal = "TRIGGERED" if price <= buy_target_price else "WAIT"

    notes: list[str] = []
    if quote is None:
        notes.append("quote_missing")
    if profile is None:
        notes.append("profile_missing")
    if annual_dividend is None:
        notes.append("annual_dividend_missing")
    if avg_yield_5y_pct is None:
        notes.append("avg_5y_yield_missing")
    elif len(yields_5y) < 5:
        notes.append(f"avg_5y_yield_points={len(yields_5y)}")

    row: dict[str, Any] = {
        "ticker": ticker,
        "signal": signal,
        "price": round(price, 2) if price is not None else None,
        "annual_dividend_per_share": round(annual_dividend, 4)
        if annual_dividend is not None
        else None,
        "current_yield_pct": current_yield_pct,
        "avg_5y_yield_pct": avg_yield_5y_pct,
        "alpha_pp": round(alpha_pp, 2),
        "target_yield_pct": target_yield_pct,
        "buy_target_price": buy_target_price,
        "drop_needed_pct": drop_needed_pct,
        "yield_observation_count": len(yields_5y),
        "notes": notes,
    }

    # WS-1: attach the dividend-basis breakdown + Step-1 decision.
    if basis is not None:
        row["dividend_basis"] = {
            "status": basis.status,
            "cadence": basis.cadence,
            "latest_declared_annualized": basis.latest_declared_annualized,
            "regular_annual_dividend": basis.regular_annual_dividend,
            "ttm_dividend_incl_special": basis.ttm_dividend_incl_special,
            "regular_forward_yield_pct": basis.regular_forward_yield_pct,
            "ttm_yield_pct": basis.ttm_yield_pct,
            "special_dividend_flag": basis.special_dividend_flag,
            "variable_policy_flag": basis.variable_policy_flag,
            "cut_flag": basis.cut_flag,
            "freeze_flag": basis.freeze_flag,
            "suspension_flag": basis.suspension_flag,
            "last_increase_date": basis.last_increase_date,
            "dividend_dates_used": basis.dividend_dates_used,
            "floor_borderline": basis.floor_borderline,
            "reasons": basis.reasons,
        }
        if floor_pct is not None:
            verdict, reason = step1_decision(
                basis, floor_pct, source_confirmed=basis.latest_declared_confirmed
            )
            row["step1_verdict"] = verdict
            row["step1_reason"] = reason
        for flag in (
            "special_dividend_flag",
            "variable_policy_flag",
            "cut_flag",
            "freeze_flag",
            "suspension_flag",
        ):
            if getattr(basis, flag):
                notes.append(flag)
        if basis.floor_borderline:
            notes.append("floor_borderline")

    pre_order_blockers: list[str] = []
    if basis is not None:
        # freeze_flag is intentionally NOT a blocker here: its disposition
        # (CONDITIONAL-PASS vs HOLD-REVIEW) is decided by synthesize_verdict
        # from step1_verdict + safety. cut/variable/suspension already force
        # step1 FAIL; listing them keeps the audit trail explicit.
        for flag in ("variable_policy_flag", "cut_flag", "suspension_flag"):
            if getattr(basis, flag):
                pre_order_blockers.append(flag)
        if basis.floor_borderline:
            pre_order_blockers.append("dividend_source_stale")

    # WS-3 FIRST (6th-review High2/High3): a missing scan is SKIPPED, never
    # silently clean; the pessimistic cap must run on every row, and a
    # completed merger must reach WS-2 *before* payout safety is assessed.
    if event_scan is None:
        event_scan = ScanResult(ticker=ticker, result="SKIPPED", scanned_at=as_of)
    triggered = str(signal) == "TRIGGERED"
    cap = apply_event_cap(event_scan, step5_triggered=triggered)
    row["event_scan"] = {
        "result": event_scan.result,
        "pending_mna": event_scan.pending_mna,
        "completed_mna_within_4q": event_scan.completed_mna_within_4q,
        "sources": event_scan.sources,
        "scanned_at": event_scan.scanned_at,
        "reasons": event_scan.reasons,
    }
    if cap["verdict_cap"]:
        row["verdict_cap"] = cap["verdict_cap"]
    row["t1_blocked"] = cap["t1_blocked"]
    # Event-scan T1/order-gate blockers are kept on the row for the order
    # gate, but excluded from the verdict-affecting set (their verdict
    # impact is already represented via verdict_cap + event_t1_blocked).
    event_order_blockers = list(cap["blockers"])
    for r in cap["reasons"]:
        notes.append(r)
    if event_scan.completed_mna_within_4q:
        notes.append("completed_merger_within_4q")

    # WS-2: sector-aware payout-safety triad. completed_merger_within_4q now
    # flows from the event scan (High3) so FITB/Comerica GAAP distortion is
    # caught from the real path, not only when financials carries the flag.
    if financials is not None:
        safety = assess_payout_safety(
            sector=financials.get("sector") or (profile or {}).get("sector"),
            annual_dividend=annual_dividend,
            gaap_eps=financials.get("gaap_eps"),
            adjusted_eps=financials.get("adjusted_eps"),
            adjusted_eps_source=financials.get("adjusted_eps_source", "UNAVAILABLE"),
            fcf_per_share=financials.get("fcf_per_share"),
            completed_merger_within_4q=bool(
                financials.get("completed_merger_within_4q", False)
                or event_scan.completed_mna_within_4q
            ),
            bank_metrics=financials.get("bank_metrics"),
            utility_metrics=financials.get("utility_metrics"),
            insurer_metrics=financials.get("insurer_metrics"),
        )
        row["payout_safety"] = {
            "sector_kind": safety.sector_kind,
            "safety_verdict": safety.safety_verdict,
            "gaap_eps_payout": safety.gaap_eps_payout,
            "adjusted_eps_payout": safety.adjusted_eps_payout,
            "fcf_payout": safety.fcf_payout,
            "adjusted_eps_source": safety.adjusted_eps_source,
            "gaap_adj_divergence": safety.gaap_adj_divergence,
            "one_off_flag": safety.one_off_flag,
            "reasons": safety.reasons,
        }
        pre_order_blockers.extend(safety.blockers)
        if safety.one_off_flag:
            notes.append("gaap_one_off")

    order_blockers = sorted(set(pre_order_blockers) | set(event_order_blockers))
    if order_blockers:
        row["pre_order_blockers"] = order_blockers

    # WS-5: synthesize the actionable verdict tier. Verdict-affecting
    # blockers exclude pure event-scan order blockers (cap handles them).
    safety_v = row.get("payout_safety", {}).get("safety_verdict")
    final = synthesize_verdict(
        step1_verdict=row.get("step1_verdict"),
        safety_verdict=safety_v,
        event_verdict_cap=row.get("verdict_cap"),
        event_t1_blocked=bool(row.get("t1_blocked", False)),
        pre_order_blockers=sorted(set(pre_order_blockers)),
    )
    row["verdict"] = final.verdict
    row["t1_blocked"] = final.t1_blocked
    row["verdict_reasons"] = final.reasons
    row["provenance"] = {
        "price_source": "fmp_quote" if quote else None,
        "dividend_source": "fmp_stock_dividend" if dividend_history else "fmp_profile_lastDiv",
        "dividend_dates_used": (basis.dividend_dates_used if basis else []),
        "payout_source": (financials or {}).get("adjusted_eps_source", "UNAVAILABLE"),
        "event_scan_result": (event_scan.result if event_scan else "NOT_SCANNED"),
        "event_scan_checked_at": (event_scan.scanned_at if event_scan else None),
        # 7th-review: audit the COMPLETE unresolved-blocker set (incl.
        # event-scan order blockers), matching row["pre_order_blockers"];
        # synthesize() intentionally consumes only the verdict-affecting
        # subset, but the audit trail must not under-report the T1 gate.
        "unresolved_blockers": order_blockers,
        "evidence_refs": [],  # populated by Claude per SKILL.md source hierarchy
    }

    return row


def render_markdown(rows: list[dict[str, Any]], as_of: str, alpha_pp: float) -> str:
    counts = {"TRIGGERED": 0, "WAIT": 0, "ASSUMPTION-REQUIRED": 0}
    verdict_counts: dict[str, int] = {}
    for row in rows:
        status = str(row.get("signal", "ASSUMPTION-REQUIRED"))
        counts[status] = counts.get(status, 0) + 1
        v = row.get("verdict")
        if v:
            verdict_counts[v] = verdict_counts.get(v, 0) + 1

    verdict_lines = [f"- {v}: `{verdict_counts[v]}`" for v in VERDICTS if v in verdict_counts]

    lines = [
        "# Kanchi Entry Signals",
        "",
        f"- as_of: `{as_of}`",
        f"- alpha_pp: `{alpha_pp:.2f}`",
        f"- ticker_count: `{len(rows)}`",
        "",
        "## Verdict Summary (WS-5 actionable tier)",
        "",
        *(verdict_lines or ["- (no verdicts; run with --yield-floor)"]),
        "",
        "## Step-5 Timing Summary",
        "",
        f"- TRIGGERED: `{counts.get('TRIGGERED', 0)}`",
        f"- WAIT: `{counts.get('WAIT', 0)}`",
        f"- ASSUMPTION-REQUIRED: `{counts.get('ASSUMPTION-REQUIRED', 0)}`",
        "",
        "## Signals",
        "",
        "> Verdict/T1-Blocked are the actionable columns. `signal` is only "
        "Step-5 timing — never act on it alone (a TRIGGERED row can be "
        "HOLD-REVIEW / T1-blocked).",
        "",
        "| Ticker | Verdict | T1 Blocked | Signal | Price | Reg Yield% | "
        "TTM Yield% | Step1 | Safety | Event Scan | Pre-order Blockers |",
        "|---|---|---|---|---:|---:|---:|---|---|---|---|",
    ]

    for row in rows:
        db = row.get("dividend_basis", {})
        ps = row.get("payout_safety", {})
        ev = row.get("event_scan", {})
        blockers = ";".join(row.get("pre_order_blockers", [])) or "-"
        lines.append(
            "| {ticker} | {verdict} | {t1} | {signal} | {price} | {ry} | "
            "{ty} | {s1} | {safety} | {ev} | {blk} |".format(
                ticker=row.get("ticker", ""),
                verdict=row.get("verdict", "-"),
                t1="YES" if row.get("t1_blocked") else "no",
                signal=row.get("signal", ""),
                price=row.get("price", ""),
                ry=db.get("regular_forward_yield_pct", ""),
                ty=db.get("ttm_yield_pct", ""),
                s1=row.get("step1_verdict", "-"),
                safety=ps.get("safety_verdict", "-"),
                ev=ev.get("result", "NOT_SCANNED"),
                blk=blockers,
            )
        )

    lines.append("")
    return "\n".join(lines)


def write_csv(rows: list[dict[str, Any]], output_path: Path) -> None:
    # Verdict/blocker columns are first-class so a CSV-only operator sees
    # the same actionable gate as the JSON (7th-review High).
    fieldnames = [
        "ticker",
        "verdict",
        "t1_blocked",
        "signal",
        "step1_verdict",
        "payout_safety_verdict",
        "event_scan_result",
        "pre_order_blockers",
        "price",
        "annual_dividend_per_share",
        "regular_forward_yield_pct",
        "ttm_yield_pct",
        "current_yield_pct",
        "avg_5y_yield_pct",
        "alpha_pp",
        "target_yield_pct",
        "buy_target_price",
        "drop_needed_pct",
        "yield_observation_count",
        "notes",
    ]

    with output_path.open("w", encoding="utf-8", newline="") as fh:
        writer = csv.DictWriter(fh, fieldnames=fieldnames, extrasaction="ignore")
        writer.writeheader()
        for row in rows:
            db = row.get("dividend_basis", {})
            ps = row.get("payout_safety", {})
            ev = row.get("event_scan", {})
            output = dict(row)
            output["notes"] = ",".join(output.get("notes", []))
            output["pre_order_blockers"] = ";".join(row.get("pre_order_blockers", []))
            output["regular_forward_yield_pct"] = db.get("regular_forward_yield_pct")
            output["ttm_yield_pct"] = db.get("ttm_yield_pct")
            output["payout_safety_verdict"] = ps.get("safety_verdict")
            output["event_scan_result"] = ev.get("result", "NOT_SCANNED")
            writer.writerow(output)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Build Kanchi Step 5 entry signals from FMP data.")
    parser.add_argument("--input", default=None, help="Path to JSON file containing tickers.")
    parser.add_argument("--tickers", default=None, help="Comma-separated ticker list.")
    parser.add_argument(
        "--alpha-pp",
        type=float,
        default=0.5,
        help="Yield alpha in percentage points (default: 0.5).",
    )
    parser.add_argument("--output-dir", default="reports", help="Directory for outputs.")
    parser.add_argument(
        "--as-of",
        default=date.today().isoformat(),
        help="As-of date (YYYY-MM-DD).",
    )
    parser.add_argument(
        "--filename-prefix",
        default="kanchi_entry_signals",
        help="Output filename prefix.",
    )
    parser.add_argument(
        "--sleep-seconds",
        type=float,
        default=0.15,
        help="Per-request wait time to reduce API throttling.",
    )
    parser.add_argument(
        "--yield-floor",
        type=float,
        default=None,
        help="Step-1 yield floor %% (e.g. 4.0 income-now, 3.0 balanced). "
        "Enables WS-1 Step-1 verdict + Data Freshness Gate.",
    )
    parser.add_argument(
        "--events-json",
        default=None,
        help="Path to a curated corporate-events JSON (WS-3 Step 4b, populated "
        "via WebSearch per SKILL.md). Absent --events-json -> SKIPPED; "
        "unknown ticker inside the events JSON -> NO_EVENT_FOUND. Both are "
        "pessimistic for TRIGGERED names (cap to HOLD-REVIEW + T1 blocked).",
    )
    parser.add_argument("--profile", default=None, help="income-now | balanced | growth-first")
    parser.add_argument("--safety-bias", default=None, help="tight | medium")
    parser.add_argument("--universe-source", default=None, help="Provenance: universe origin.")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    tickers = load_tickers(Path(args.input) if args.input else None, args.tickers)
    if not tickers:
        raise SystemExit("No tickers provided. Use --tickers or --input.")

    api_key = os.getenv("FMP_API_KEY")
    if not api_key:
        raise SystemExit("FMP_API_KEY is not set.")

    client = FMPClient(api_key=api_key, sleep_seconds=args.sleep_seconds)

    quotes = client.get_batch_quotes(tickers)
    profiles = client.get_batch_profiles(tickers)

    scanner = None
    if args.events_json:
        from event_scanner import ManualEventScanner

        scanner = ManualEventScanner(args.events_json)

    rows: list[dict[str, Any]] = []
    for ticker in tickers:
        metrics = client.get_ratios(ticker, limit=10)
        dividend_history = client.get_stock_dividend(ticker)
        event_scan = scanner.scan(ticker, args.as_of) if scanner else None
        sector = (profiles.get(ticker) or {}).get("sector")
        financials = client.get_financials(ticker, sector=sector)
        row = build_entry_row(
            ticker=ticker,
            alpha_pp=args.alpha_pp,
            quote=quotes.get(ticker),
            profile=profiles.get(ticker),
            key_metrics=metrics,
            dividend_history=dividend_history,
            floor_pct=args.yield_floor,
            financials=financials,
            event_scan=event_scan,
            as_of=args.as_of,
        )
        rows.append(row)

    output_dir = Path(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    prefix = f"{args.filename_prefix}_{args.as_of}"
    json_path = output_dir / f"{prefix}.json"
    csv_path = output_dir / f"{prefix}.csv"
    md_path = output_dir / f"{prefix}.md"

    payload = {
        "schema_version": SCHEMA_VERSION,
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "as_of": args.as_of,
        "alpha_pp": args.alpha_pp,
        "yield_floor_pct": args.yield_floor,
        "run_context": build_run_context(
            profile=args.profile,
            yield_floor_pct=args.yield_floor,
            safety_bias=args.safety_bias,
            universe_source=args.universe_source,
            excluded_asset_types=None,
        ),
        "ticker_count": len(tickers),
        "api_calls": client.api_calls,
        "rows": rows,
    }
    json_path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
    write_csv(rows, csv_path)
    md_path.write_text(
        render_markdown(rows, as_of=args.as_of, alpha_pp=args.alpha_pp) + "\n", encoding="utf-8"
    )

    print(f"Wrote JSON: {json_path}")
    print(f"Wrote CSV: {csv_path}")
    print(f"Wrote MD: {md_path}")
    print(f"API calls: {client.api_calls}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Generate a deterministic Kanchi SOP planning markdown file."""

from __future__ import annotations

import argparse
import json
from datetime import date
from pathlib import Path
from typing import Any


def parse_ticker_csv(raw: str) -> list[str]:
    tickers: list[str] = []
    for part in raw.split(","):
        value = part.strip().upper()
        if not value:
            continue
        if value not in tickers:
            tickers.append(value)
    return tickers


def normalize_candidates(payload: dict[str, Any]) -> list[dict[str, str]]:
    raw = payload.get("candidates")
    if not isinstance(raw, list):
        raw = payload.get("stocks")
    if isinstance(raw, list):
        normalized: list[dict[str, str]] = []
        for item in raw:
            if isinstance(item, str):
                ticker = item.strip().upper()
                if ticker:
                    normalized.append({"ticker": ticker, "bucket": "unassigned"})
            elif isinstance(item, dict):
                ticker = str(item.get("ticker") or item.get("symbol") or "").strip().upper()
                bucket = str(item.get("bucket", "unassigned")).strip().lower() or "unassigned"
                if ticker:
                    normalized.append({"ticker": ticker, "bucket": bucket})
        return normalized
    return []


def load_candidates(
    input_path: Path | None, tickers_csv: str | None
) -> tuple[list[dict[str, str]], str]:
    if input_path:
        payload = json.loads(input_path.read_text())
        profile = str(payload.get("profile", "balanced")).strip().lower() or "balanced"
        candidates = normalize_candidates(payload)
        if candidates:
            return candidates, profile
        fallback = payload.get("tickers", [])
        if isinstance(fallback, list):
            candidates = [
                {"ticker": str(item).strip().upper(), "bucket": "unassigned"}
                for item in fallback
                if str(item).strip()
            ]
            return candidates, profile
        return [], profile

    if tickers_csv:
        return (
            [
                {"ticker": ticker, "bucket": "unassigned"}
                for ticker in parse_ticker_csv(tickers_csv)
            ],
            "balanced",
        )

    return [], "balanced"


def render_markdown(candidates: list[dict[str, str]], as_of: str, profile: str) -> str:
    lines = [
        "# Kanchi SOP Plan",
        "",
        f"- as_of: `{as_of}`",
        f"- profile: `{profile}`",
        f"- candidate_count: `{len(candidates)}`",
        "",
        "## Candidate Universe",
        "",
        "| Ticker | Bucket | Step1 | Step2 | Step3 | Step4 | Step5 |",
        "|---|---|---|---|---|---|---|",
    ]

    for item in candidates:
        lines.append(
            "| {ticker} | {bucket} | todo | todo | todo | todo | todo |".format(
                ticker=item["ticker"],
                bucket=item["bucket"],
            )
        )

    lines.extend(
        [
            "",
            "## Pullback Entry Notes",
            "",
            "- Default split order: `40% -> 30% -> 30%`",
            "- Default yield alpha: `+0.5pp` vs 5y average yield.",
            "- Add one-line invalidation per ticker before finalizing orders.",
            "",
        ]
    )
    return "\n".join(lines)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Build Kanchi SOP plan markdown.")
    parser.add_argument("--input", help="Path to JSON input with candidates/tickers.", default=None)
    parser.add_argument("--tickers", help="Comma-separated ticker list.", default=None)
    parser.add_argument("--output-dir", default="reports", help="Output directory path.")
    parser.add_argument(
        "--as-of", default=date.today().isoformat(), help="As-of date (YYYY-MM-DD)."
    )
    parser.add_argument("--filename", default=None, help="Optional output filename.")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    input_path = Path(args.input) if args.input else None
    candidates, profile = load_candidates(input_path, args.tickers)
    if not candidates:
        raise SystemExit(
            "No candidates found. Provide --tickers or --input with candidates/tickers."
        )

    output_dir = Path(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    filename = args.filename or f"kanchi_sop_plan_{args.as_of}.md"
    output_path = output_dir / filename
    output_path.write_text(
        render_markdown(candidates, args.as_of, profile) + "\n", encoding="utf-8"
    )
    print(f"Wrote SOP plan: {output_path}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""WS-1: dividend-basis engine (regular / special / variable / freeze / cut).

Pure, dependency-free, offline-testable. Replaces the old
``annual_dividend = profile.lastDiv`` shortcut in build_entry_signals.py,
which fed a trailing/TTM figure that (a) lagged the latest declared raise
(defect D5: CFR FAIL'd at 2.97% though latest declared = 3.06%) and
(b) silently included special/variable dividends (D4: ORI/CALM traps).

Detection ORDER is fixed by improvement-plan v2.1 R-1 — specials MUST be
removed before the variable-policy CoV is computed, otherwise a special
dividend (ORI) inflates the residual stream's variance and the issuer is
misclassified as a CALM-style variable payer.
"""

from __future__ import annotations

import statistics
from dataclasses import dataclass, field
from datetime import date, datetime

from thresholds import (
    DIVIDEND_EQUALITY_EPS,
    FLOOR_FRESHNESS_BAND_PP,
    FREEZE_GRACE_DAYS,
    MIN_REGULAR_PAYS,
    SPECIAL_EXTREME_MULTIPLE,
    SPECIAL_OUTLIER_MULTIPLE,
    SPECIAL_REVERT_MULTIPLE,
    VARIABLE_POLICY_COV,
)

_SPECIAL_KEYWORDS = ("special", "supplemental", "one-time", "one time", "extra")
_MAINTAIN_KEYWORDS = ("maintain", "unchanged", "same annualized", "no change", "flat")

# 4th-review point 10: a generic "% of net income" appears in ordinary bank /
# insurer payout-ratio language and must NOT alone flag a variable policy.
# Only the strong tier (an explicit variable/earnings-linked policy) flips the
# flag; the weak tier just leaves an audit note.
_STRONG_VARIABLE_KEYWORDS = (
    "variable dividend policy",
    "variable dividend",
    "one-third of net income",
    "1/3 of net income",
    "distribution based on earnings",
    "dividend equal to",
)
_WEAK_PAYOUT_KEYWORDS = (
    "target payout ratio",
    "capital return framework",
    "dividend payout ratio",
    "% of net income",
    "percent of net income",
)

_CADENCE_BY_GAP_DAYS = (
    (45, "monthly", 12),
    (135, "quarterly", 4),
    (240, "semiannual", 2),
    (450, "annual", 1),
)


def _parse_date(value: str) -> date | None:
    for fmt in ("%Y-%m-%d", "%Y/%m/%d"):
        try:
            return datetime.strptime(value, fmt).date()
        except (ValueError, TypeError):
            continue
    return None


@dataclass
class DividendBasis:
    status: str  # "OK" | "ASSUMPTION-REQUIRED" | "NO-DIVIDEND"
    reasons: list[str] = field(default_factory=list)
    cadence: str = "irregular"
    pays_per_year: int | None = None
    latest_declared_dividend: float | None = None
    latest_declared_annualized: float | None = None
    regular_annual_dividend: float | None = None
    ttm_dividend_incl_special: float | None = None
    special_dividend_flag: bool = False
    variable_policy_flag: bool = False
    cut_flag: bool = False
    freeze_flag: bool = False
    suspension_flag: bool = False
    latest_declared_confirmed: bool = False
    last_increase_date: str | None = None
    dividend_dates_used: list[str] = field(default_factory=list)
    regular_forward_yield_pct: float | None = None
    ttm_yield_pct: float | None = None
    floor_borderline: bool = False


def _infer_cadence(sorted_dates: list[date]) -> tuple[str, int | None]:
    if len(sorted_dates) < 2:
        return "irregular", None
    gaps = [
        (sorted_dates[i + 1] - sorted_dates[i]).days
        for i in range(len(sorted_dates) - 1)
        if (sorted_dates[i + 1] - sorted_dates[i]).days > 0
    ]
    if not gaps:
        return "irregular", None
    median_gap = statistics.median(gaps)
    for max_gap, name, ppy in _CADENCE_BY_GAP_DAYS:
        if median_gap <= max_gap:
            # High dispersion vs the cadence => irregular schedule.
            if len(gaps) >= 3 and statistics.pstdev(gaps) > median_gap * 0.6:
                return "irregular", ppy
            return name, ppy
    return "irregular", 1


def _is_explicit_special(label: object) -> bool:
    if not isinstance(label, str):
        return False
    low = label.lower()
    return any(k in low for k in _SPECIAL_KEYWORDS)


def _sum_window(pays: list[tuple[date, float]], end: date, start_days: int, end_days: int) -> float:
    """Sum amounts with (end - start_days) <= d < (end - end_days) days old."""
    total = 0.0
    for d, amt in pays:
        age = (end - d).days
        if end_days <= age < start_days:
            total += amt
    return total


def analyze_dividends(
    history: list[dict],
    price: float | None,
    *,
    issuer_language: str | None = None,
    floor_pct: float | None = None,
    as_of_date: str | None = None,
) -> DividendBasis:
    """Analyze a stock_dividend `historical` list into a DividendBasis.

    `history` items: {"date": "YYYY-MM-DD", "dividend": float, "label": str?}.
    """
    rows: list[tuple[date, float, object]] = []
    declared: dict[date, bool] = {}
    for item in history or []:
        d = _parse_date(str(item.get("date", "")))
        amt = item.get("dividend")
        try:
            amt = float(amt)
        except (TypeError, ValueError):
            amt = None
        if d is None or amt is None or amt <= 0:
            continue
        rows.append((d, amt, item.get("label")))
        # A non-empty declarationDate means the board formally declared
        # this dividend -> authoritative confirmation for the Data
        # Freshness Gate (5th-review #4 confirmed-source path).
        decl = item.get("declarationDate")
        declared[d] = bool(decl and str(decl).strip())

    if not rows:
        return DividendBasis(status="NO-DIVIDEND", reasons=["no_dividend_history"])

    rows.sort(key=lambda r: r[0])
    all_dates = [r[0] for r in rows]
    cadence, ppy = _infer_cadence(all_dates)

    # --- Step 2: explicit special labels/language ---
    explicit_special = {i for i, r in enumerate(rows) if _is_explicit_special(r[2])}

    # --- Step 3: amount-outliers vs the TRAILING-local median of prior
    #     regular pays (v2.1 R-1). A global median over a 10y+ history is
    #     dragged down by old small dividends, so ordinary long-term
    #     dividend growth would be misclassified as a special and
    #     latest_declared would collapse to a stale amount (D4 recurrence).
    #     Compare each pay only against its recent prior regular pays.
    outlier_special: set[int] = set()
    _TRAILING = max(ppy or 4, 8)
    for i, r in enumerate(rows):
        if i in explicit_special:
            continue
        prior_regular = [
            rows[j][1]
            for j in range(max(0, i - _TRAILING), i)
            if j not in explicit_special and j not in outlier_special
        ]
        if len(prior_regular) < MIN_REGULAR_PAYS:
            # Too little local history to call an outlier -> treat as
            # regular (conservative; avoids flagging early history).
            continue
        local_med = statistics.median(prior_regular)
        if local_med <= 0:
            continue
        if r[1] > SPECIAL_EXTREME_MULTIPLE * local_med:
            outlier_special.add(i)  # extreme isolated spike (magnitude)
        elif r[1] > SPECIAL_OUTLIER_MULTIPLE * local_med:
            # One-off only if a later pay reverts toward the local median;
            # ordinary steep growth keeps rising and is NOT a special.
            revert = any(
                rows[k][1] <= SPECIAL_REVERT_MULTIPLE * local_med
                for k in range(i + 1, min(len(rows), i + 3))
                if k not in explicit_special
            )
            if revert:
                outlier_special.add(i)

    special_idx = explicit_special | outlier_special
    special_flag = bool(special_idx)

    # --- Step 4: residual = regular stream ---
    regular = [(r[0], r[1]) for i, r in enumerate(rows) if i not in special_idx]
    all_pays = [(r[0], r[1]) for r in rows]

    reasons: list[str] = []
    if special_flag:
        reasons.append(f"special_dividends_excluded={len(special_idx)}")

    if len(regular) < MIN_REGULAR_PAYS:
        return DividendBasis(
            status="ASSUMPTION-REQUIRED",
            reasons=reasons + [f"insufficient_regular_pays={len(regular)}"],
            cadence=cadence,
            pays_per_year=ppy,
            special_dividend_flag=special_flag,
        )

    latest_date, latest_amt = regular[-1]
    latest_annualized = latest_amt * ppy if ppy else None

    asof = all_dates[-1]
    asof_reg = regular[-1][0]
    regular_ttm = _sum_window(regular, asof, 366, 0)
    incl_special_ttm = _sum_window(all_pays, asof, 366, 0)

    # --- Step 5: variable-policy on the *residual* (post-special) stream ---
    window = [a for _, a in regular[-max(ppy or 4, 8) :]]
    variable_flag = False
    if len(window) >= MIN_REGULAR_PAYS:
        mean = statistics.fmean(window)
        if mean > 0:
            cov = statistics.pstdev(window) / mean
            variable_flag = cov > VARIABLE_POLICY_COV
    if issuer_language:
        low = issuer_language.lower()
        if any(k in low for k in _STRONG_VARIABLE_KEYWORDS):
            variable_flag = True
        elif not variable_flag and any(k in low for k in _WEAK_PAYOUT_KEYWORDS):
            reasons.append("weak_payout_language_only_review_policy")

    # --- last increase date (ascending scan over regular amounts) ---
    last_increase: date | None = None
    for i in range(1, len(regular)):
        if regular[i][1] > regular[i - 1][1] + DIVIDEND_EQUALITY_EPS:
            last_increase = regular[i][0]

    # --- freeze / cut via year-over-year per-pay RATE (frequency-independent,
    #     robust to rolling-window pay-count drift; v2.1 R-2) ---
    eps = DIVIDEND_EQUALITY_EPS
    cut_flag = False
    freeze_flag = False
    cur_rate = regular[-1][1]
    prior_idx = len(regular) - 1 - ppy if ppy else -1
    prior_rate = regular[prior_idx][1] if prior_idx >= 0 else None
    lang_maintains = bool(
        issuer_language and any(k in issuer_language.lower() for k in _MAINTAIN_KEYWORDS)
    )
    if prior_rate is not None:
        if cur_rate < prior_rate - eps:
            cut_flag = True
        elif abs(cur_rate - prior_rate) <= eps:
            cadence_days = {"monthly": 30, "quarterly": 91, "semiannual": 182}.get(cadence, 365)
            days_since_increase = (asof_reg - last_increase).days if last_increase else 10_000
            time_confirms = days_since_increase > cadence_days + FREEZE_GRACE_DAYS
            if time_confirms or lang_maintains:
                freeze_flag = True
            else:
                reasons.append("flat_window_pending_next_raise")
    else:
        reasons.append("no_prior_year_rate")
        if lang_maintains:
            freeze_flag = True

    # --- suspension: an expected regular declaration is overdue (4th-review
    #     point 4). Distinct from cut (rate down) and freeze (rate held). ---
    cadence_days = {"monthly": 30, "quarterly": 91, "semiannual": 182, "annual": 365}.get(
        cadence, 365
    )
    suspension_flag = False
    ref_date = _parse_date(str(as_of_date)) if as_of_date else None
    if ref_date is not None:
        days_since_last_pay = (ref_date - asof_reg).days
        if days_since_last_pay > cadence_days + FREEZE_GRACE_DAYS:
            suspension_flag = True
            reasons.append(f"declaration_overdue_{days_since_last_pay}d")

    fwd_yield = (
        round(latest_annualized / price * 100, 2)
        if latest_annualized is not None and price and price > 0
        else None
    )
    ttm_yield = (
        round(incl_special_ttm / price * 100, 2)
        if price and price > 0 and incl_special_ttm
        else None
    )

    floor_borderline = False
    if floor_pct is not None and fwd_yield is not None:
        if abs(fwd_yield - floor_pct) <= FLOOR_FRESHNESS_BAND_PP:
            floor_borderline = True
            reasons.append(f"floor_borderline_within_{FLOOR_FRESHNESS_BAND_PP}pp")

    return DividendBasis(
        status="OK",
        reasons=reasons,
        cadence=cadence,
        pays_per_year=ppy,
        latest_declared_dividend=round(latest_amt, 4),
        latest_declared_annualized=round(latest_annualized, 4) if latest_annualized else None,
        regular_annual_dividend=round(regular_ttm, 4) if regular_ttm else None,
        ttm_dividend_incl_special=round(incl_special_ttm, 4) if incl_special_ttm else None,
        special_dividend_flag=special_flag,
        variable_policy_flag=variable_flag,
        cut_flag=cut_flag,
        freeze_flag=freeze_flag,
        suspension_flag=suspension_flag,
        latest_declared_confirmed=declared.get(regular[-1][0], False),
        last_increase_date=last_increase.isoformat() if last_increase else None,
        dividend_dates_used=[d.isoformat() for d, _ in regular[-8:]],
        regular_forward_yield_pct=fwd_yield,
        ttm_yield_pct=ttm_yield,
        floor_borderline=floor_borderline,
    )


def step1_decision(
    basis: DividendBasis,
    floor_pct: float,
    *,
    source_confirmed: bool = False,
) -> tuple[str, str]:
    """Map a DividendBasis to a Step-1 verdict per the v2 decision matrix.

    Returns (verdict, reason). Verdicts: FAIL / STEP1-RECHECK / HOLD-REVIEW /
    STEP1-PASS (caller refines STEP1-PASS into the full tier via WS-2/WS-5).
    """
    if basis.status == "NO-DIVIDEND":
        return "FAIL", "no_dividend"
    if basis.status == "ASSUMPTION-REQUIRED":
        return "STEP1-RECHECK", ";".join(basis.reasons) or "assumption_required"
    if basis.suspension_flag:
        return "FAIL", "dividend_suspension_suspected"
    if basis.variable_policy_flag:
        return "FAIL", "variable_dividend_policy"
    if basis.cut_flag:
        return "FAIL", "dividend_cut"

    fwd = basis.regular_forward_yield_pct
    if fwd is None:
        return "STEP1-RECHECK", "regular_yield_unavailable"

    if fwd < floor_pct:
        # Data Freshness Gate: near-floor + unconfirmed source -> recheck,
        # never a hard FAIL (defect D5).
        if basis.floor_borderline and not source_confirmed:
            return "STEP1-RECHECK", "floor_borderline_unconfirmed_dividend"
        return "FAIL", f"regular_yield_{fwd}_below_floor_{floor_pct}"

    # At/above floor on regular yield.
    if basis.floor_borderline and not source_confirmed:
        return "STEP1-RECHECK", "floor_borderline_confirm_latest_declared"
    if basis.freeze_flag:
        return "HOLD-REVIEW", "dividend_freeze"
    return "STEP1-PASS", "regular_yield_above_floor"

"""WS-3: forward/recent corporate-action scanner (isolated, deterministic-testable).

Critical-review CR-2: the SOP core is a *deterministic* workflow. WebSearch
is non-deterministic and unavailable in some environments, so the event
layer is ISOLATED behind a small interface:

- The pure logic here (materiality gate, classification, pessimistic cap)
  never calls the network and is fully unit-testable.
- A scanner is INJECTED. `ManualEventScanner` reads a curated JSON events
  file (populated by Claude via WebSearch following SKILL.md Step 4b, with
  the source hierarchy IR > SEC > exchange > wire > portal). CI injects a
  fixture scanner -- real WebSearch is never called in tests.

Closes D3 (MKC-Unilever mega-merger missed: Step 4 was backward-only).

4th-review points folded in:
  #5  result in {FAILED-DEGRADED, SKIPPED, NO_EVENT_FOUND} AND Step-5
      timing TRIGGERED  ->  verdict cap HOLD-REVIEW + T1 BLOCKED
      (not merely a provenance stamp).
  #7  sector-specific materiality + rolling-24m cumulative M&A.
  #11 CLEAN_CONFIRMED (primary source checked) is stronger than
      NO_EVENT_FOUND (search only); the latter is treated pessimistically.
"""

from __future__ import annotations

import json
from dataclasses import dataclass, field
from pathlib import Path
from typing import Protocol

from thresholds import (
    MNA_LEVERAGE_DELTA_EBITDA,
    MNA_SHARE_ISSUANCE_PCT,
    MNA_TX_VALUE_PCT_MCAP,
)

# event_scan_result enum (4th-review #11)
CLEAN_CONFIRMED = "CLEAN_CONFIRMED"
NO_EVENT_FOUND = "NO_EVENT_FOUND"
MAJOR_EVENT = "MAJOR_EVENT"
MINOR_EVENT_CAUTION = "MINOR_EVENT_CAUTION"
FAILED_DEGRADED = "FAILED-DEGRADED"
SKIPPED = "SKIPPED"

# Results that, on a TRIGGERED Step-5 name, must pessimistically cap.
_PESSIMISTIC_RESULTS = {FAILED_DEGRADED, SKIPPED, NO_EVENT_FOUND, MAJOR_EVENT}

# The complete valid enum. Anything else (e.g. a typo "FAILED_DEGRADED"
# instead of "FAILED-DEGRADED") is coerced to FAILED_DEGRADED — default-deny,
# never silently clean (6th-review Med1).
_VALID_RESULTS = {
    CLEAN_CONFIRMED,
    NO_EVENT_FOUND,
    MAJOR_EVENT,
    MINOR_EVENT_CAUTION,
    FAILED_DEGRADED,
    SKIPPED,
}


def _coerce_result(value: object) -> str:
    s = str(value).strip()
    return s if s in _VALID_RESULTS else FAILED_DEGRADED


_ROLLING_24M_TX_PCT_MCAP = 0.15


@dataclass
class ScanResult:
    ticker: str
    result: str
    pending_mna: bool = False
    completed_mna_within_4q: bool = False
    spinoff: bool = False
    dividend_policy_change: bool = False
    rating_action: bool = False
    sources: list[str] = field(default_factory=list)
    scanned_at: str | None = None
    reasons: list[str] = field(default_factory=list)


class EventScanner(Protocol):
    def scan(self, ticker: str, as_of: str) -> ScanResult: ...


def is_major_structural_event(deal: dict) -> tuple[bool, list[str]]:
    """Generic + sector-specific + rolling-cumulative materiality (v2.1 R-4, #7).

    `deal` keys (all optional): tx_value, market_cap, share_issuance_pct,
    leverage_delta_ebitda, control_change, listing_change, hq_change,
    structure (merger_of_equals|reverse_morris_trust|spinoff|large_asset_sale),
    policy_change (dividend|rating|leverage), sector, sector_metrics,
    rolling_24m_tx_value.
    """
    reasons: list[str] = []
    mcap = deal.get("market_cap") or 0
    tx = deal.get("tx_value") or 0

    if mcap and tx and tx / mcap > MNA_TX_VALUE_PCT_MCAP:
        reasons.append(f"tx_value_{round(tx / mcap * 100)}pct_mcap")
    if (deal.get("share_issuance_pct") or 0) > MNA_SHARE_ISSUANCE_PCT * 100:
        reasons.append("share_issuance_gt_10pct")
    if (deal.get("leverage_delta_ebitda") or 0) > MNA_LEVERAGE_DELTA_EBITDA:
        reasons.append("leverage_delta_gt_0.5x_ebitda")
    if deal.get("control_change") or deal.get("listing_change") or deal.get("hq_change"):
        reasons.append("control_or_listing_or_hq_change")
    if deal.get("structure") in (
        "merger_of_equals",
        "reverse_morris_trust",
        "spinoff",
        "large_asset_sale",
    ):
        reasons.append(f"structure_{deal['structure']}")
    if deal.get("policy_change") in ("dividend", "rating", "leverage"):
        reasons.append(f"policy_change_{deal['policy_change']}")

    # Sector-specific materiality (#7).
    sm = deal.get("sector_metrics") or {}
    sector = (deal.get("sector") or "").lower()
    if sector in ("bank", "banks", "financial", "financials", "financial services"):
        if (sm.get("acquired_assets_pct") or 0) > 10 or (sm.get("acquired_deposits_pct") or 0) > 10:
            reasons.append("bank_acquired_assets_or_deposits_gt_10pct")
        if sm.get("integration_charge_material"):
            reasons.append("bank_integration_charge_material")
    elif sector in ("utility", "utilities"):
        if (sm.get("acq_or_capex_pct_rate_base") or 0) > 10:
            reasons.append("utility_acq_capex_gt_10pct_rate_base")
        if sm.get("equity_issuance_required"):
            reasons.append("utility_equity_issuance_required")
        if sm.get("regulatory_approval_pending"):
            reasons.append("utility_regulatory_approval_pending")
    elif sector in ("insurer", "insurers", "insurance"):
        if sm.get("affects_statutory_or_reserves_or_combined_ratio"):
            reasons.append("insurer_capital_reserves_combined_ratio_impact")

    # Rolling 24-month cumulative M&A (#7).
    if mcap and (deal.get("rolling_24m_tx_value") or 0) / mcap > _ROLLING_24M_TX_PCT_MCAP:
        reasons.append("rolling_24m_tx_gt_15pct_mcap")

    return (bool(reasons), reasons)


def apply_event_cap(scan: ScanResult, *, step5_triggered: bool) -> dict:
    """Pessimistic cap (user decision + 4th-review #5).

    Returns {verdict_cap, t1_blocked, blockers, reasons}. verdict_cap is None
    when no cap applies (caller keeps the computed verdict).
    """
    blockers: list[str] = []
    reasons: list[str] = []

    # Default-deny: an unknown/typo result is coerced to FAILED-DEGRADED so
    # the pessimistic path runs instead of falling through to clean (Med1).
    result = _coerce_result(scan.result)

    if result == MAJOR_EVENT:
        blockers.append("major_structural_event")
        reasons.extend(scan.reasons or ["major_structural_event"])
        return {
            "verdict_cap": "HOLD-REVIEW",
            "t1_blocked": True,
            "blockers": blockers,
            "reasons": reasons,
        }

    if result in (FAILED_DEGRADED, SKIPPED):
        blockers.append("event_scan_failed_or_skipped")
        reasons.append(f"event_scan_{result}")
    elif result == NO_EVENT_FOUND:
        blockers.append("event_scan_primary_source_unconfirmed")
        reasons.append("no_event_found_primary_source_unchecked")
    elif result == MINOR_EVENT_CAUTION:
        return {
            "verdict_cap": None,
            "t1_blocked": False,
            "blockers": [],
            "reasons": ["minor_bolt_on_event_caution_note"],
        }
    else:  # exactly CLEAN_CONFIRMED (unknowns were coerced away above)
        return {"verdict_cap": None, "t1_blocked": False, "blockers": [], "reasons": []}

    # Weak/failed scans only HARD-cap when the name is actually entry-ready.
    if step5_triggered:
        return {
            "verdict_cap": "HOLD-REVIEW",
            "t1_blocked": True,
            "blockers": blockers,
            "reasons": reasons + ["pessimistic_cap_triggered_name"],
        }
    return {
        "verdict_cap": None,
        "t1_blocked": True,
        "blockers": blockers,
        "reasons": reasons,
    }


class ManualEventScanner:
    """Reads a curated events JSON: {ticker: {result, ...ScanResult fields}}.

    Populated by Claude per SKILL.md Step 4b (WebSearch + IR/SEC primary
    sources). Unknown tickers -> NO_EVENT_FOUND (pessimistic, not CLEAN).
    """

    def __init__(self, events_path: str | Path):
        self._events: dict = {}
        p = Path(events_path)
        if p.exists():
            data = json.loads(p.read_text())
            if isinstance(data, dict):
                self._events = data.get("events", data)

    def scan(self, ticker: str, as_of: str) -> ScanResult:
        rec = self._events.get(ticker.upper())
        if not isinstance(rec, dict):
            return ScanResult(
                ticker=ticker.upper(),
                result=NO_EVENT_FOUND,
                scanned_at=as_of,
                reasons=["ticker_not_in_events_file"],
            )
        return ScanResult(
            ticker=ticker.upper(),
            result=_coerce_result(rec.get("result", NO_EVENT_FOUND)),
            pending_mna=bool(rec.get("pending_mna", False)),
            completed_mna_within_4q=bool(rec.get("completed_mna_within_4q", False)),
            spinoff=bool(rec.get("spinoff", False)),
            dividend_policy_change=bool(rec.get("dividend_policy_change", False)),
            rating_action=bool(rec.get("rating_action", False)),
            sources=list(rec.get("sources", [])),
            scanned_at=str(rec.get("scanned_at", as_of)),
            reasons=list(rec.get("reasons", [])),
        )


class FixtureEventScanner:
    """Test/double scanner: constructed from an in-memory dict."""

    def __init__(self, events: dict[str, ScanResult]):
        self._events = events

    def scan(self, ticker: str, as_of: str) -> ScanResult:
        return self._events.get(
            ticker.upper(),
            ScanResult(ticker=ticker.upper(), result=NO_EVENT_FOUND, scanned_at=as_of),
        )

#!/usr/bin/env python3
# GENERATED by scripts/generate_fmp_client.py — do not edit.
# Source of truth: scripts/fmp_client/ (core_template.py.tmpl, registry.py, extensions/).
# Regenerate: python3 scripts/generate_fmp_client.py
"""
FMP API Client for kanchi-dividend-sop

Provides rate-limited access to Financial Modeling Prep API endpoints.

Features:
- Rate limiting (0.3s between requests)
- Automatic retry on 429 errors
- Session caching for duplicate requests
- Stable-only dividend/value routing and response aliases
"""

import os
import re
import sys
import time
from datetime import datetime, timedelta
from typing import Optional
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

try:
    import requests
except ImportError:
    print("ERROR: requests library not found. Install with: pip install requests", file=sys.stderr)
    sys.exit(1)

# Mirror of scripts/provider_contracts.py::redact_url (clients are standalone).
# Requires a `?`/`&` prefix, so this masks apikey=/api_key= in URL query
# strings, not a rendered params dict (unreachable today: only
# RequestException text -- which can embed the full request URL -- is
# ever printed here, never a bare params repr()).
_APIKEY_RE = re.compile(r"([?&](?:apikey|api_key)=)[^&\s]+", re.IGNORECASE)


def _redact_key(text: str) -> str:
    """Mask apikey=/api_key= query values so stderr never carries the key."""
    return _APIKEY_RE.sub(r"\1REDACTED", str(text))


# --- FMP endpoint fallback: stable (new users) -> v3 (legacy users) ---


def _et_today(now=None):
    """America/New_York calendar date for provider query windows (issue #427).

    An explicit aware ``now`` wins (deterministic runs/tests); otherwise the
    current instant is converted to America/New_York. ``now`` must be
    timezone-aware when supplied. Fail-closed: missing TZ data raises
    RuntimeError naming the tzdata requirement instead of silently falling
    back to the runner's local clock.
    """
    try:
        tz = ZoneInfo("America/New_York")
    except ZoneInfoNotFoundError as exc:
        raise RuntimeError("America/New_York timezone data is missing; pip install tzdata") from exc
    if now is not None:
        if now.tzinfo is None:
            raise ValueError("now must be timezone-aware")
        return now.astimezone(tz).date()
    return datetime.now(tz).date()


def _stable_hist_url(base, symbols_str, params):
    """stable/historical-price-eod/full?symbol=^GSPC&from=...&to=..."""
    params["symbol"] = symbols_str
    # New stable EOD endpoint ignores `timeseries`; convert to from/to range
    # to bound the payload. Use 2x calendar days to cover N trading days
    # (trading-day/calendar-day ratio ~252/365 ~0.69, so *2 leaves headroom).
    days = params.pop("timeseries", None)
    if days is not None:
        today = _et_today()
        params["from"] = (today - timedelta(days=int(days) * 2)).isoformat()
        params["to"] = today.isoformat()
    return base, params


def _v3_hist_url(base, symbols_str, params):
    """api/v3/historical-price-full/^GSPC?timeseries=80"""
    return f"{base}/{symbols_str}", params


_FMP_ENDPOINTS = {
    "historical": [
        ("https://financialmodelingprep.com/stable/historical-price-eod/full", _stable_hist_url),
        ("https://financialmodelingprep.com/api/v3/historical-price-full", _v3_hist_url),
    ],
}


def _normalize_eod_flat_list(data, symbols_str: str, limit: Optional[int] = None):
    """Convert stable/historical-price-eod/full flat list to v3-compatible dict.

    Input  : [{"symbol": "SPY", "date": "...", "open": ..., ...}, ...]
    Output : {"symbol": "SPY", "historical": [{"date": ..., "open": ..., ...}, ...]}

    Returns the input unchanged if not a list (passthrough for v3 dict /
    historicalStockList responses). Returns None when no row matches the
    requested symbol; the caller will record the failure and try the next
    endpoint.

    If `limit` is provided (the original `timeseries=N` request), the
    `historical` list is truncated to the first `limit` entries. The new
    EOD endpoint ignores `timeseries` and returns the full available history,
    so the caller's date-range bounding plus this truncation together preserve
    the legacy "most-recent N rows" contract. Truncation assumes descending
    date order, which the FMP EOD endpoint provides (verified live).

    Note: empty list ``[]`` does not reach this normalizer because the caller's
    ``if not data: continue`` falsy check handles it earlier in
    ``_request_with_fallback``.
    """
    if not isinstance(data, list):
        return data
    if not data:
        return None
    norm_target = symbols_str.replace("-", ".")
    matched_symbol = None
    historical = []
    for row in data:
        if not isinstance(row, dict):
            continue
        # Be permissive: single-symbol endpoint may omit per-row "symbol".
        # Treat missing symbol as belonging to the requested symbols_str.
        row_sym = row.get("symbol") or symbols_str
        if row_sym.replace("-", ".") != norm_target:
            continue
        matched_symbol = matched_symbol or row_sym
        historical.append({k: v for k, v in row.items() if k != "symbol"})
    if not historical:
        return None
    if limit is not None and limit > 0:
        historical = historical[:limit]
    return {"symbol": matched_symbol or symbols_str, "historical": historical}


class FMPClient:
    """Client for Financial Modeling Prep API with rate limiting and caching"""

    BASE_URL = "https://financialmodelingprep.com/api/v3"
    RATE_LIMIT_DELAY = 0.3  # 300ms between requests

    _ENDPOINT_FAILURE_THRESHOLD = 3  # disable endpoint after N consecutive failures
    STABLE_URL = "https://financialmodelingprep.com/stable"

    def __init__(self, api_key: Optional[str] = None):
        self.api_key = (api_key or os.getenv("FMP_API_KEY") or "").strip()
        if not self.api_key:
            raise ValueError(
                "FMP API key required. Set FMP_API_KEY environment variable "
                "or pass api_key parameter."
            )
        self.session = requests.Session()
        self.session.headers.update({"apikey": self.api_key})
        self.timeout = 30
        self.cache = {}
        self.last_call_time = 0
        self.rate_limit_reached = False
        self.retry_count = 0
        self.max_retries = 1
        self.api_calls_made = 0
        # Circuit breaker: track consecutive failures per endpoint URL prefix
        self._endpoint_failures: dict[str, int] = {}
        self._disabled_endpoints: set[str] = set()
        # Most recent transport-level failure reason; set by _rate_limited_get
        # so _request_with_fallback can surface suppressed errors even when
        # an endpoint was called with quiet=True.
        self._last_error: Optional[str] = None

    def _redact_secret(self, text) -> str:
        """Mask raw and URL-encoded key echoes before diagnostics are truncated."""
        from urllib.parse import quote, quote_plus

        text = _redact_key(text)
        if not self.api_key:
            return text
        patterns = {re.escape(self.api_key)}
        for encoded in (quote(self.api_key, safe=""), quote_plus(self.api_key, safe="")):
            # Percent triplets are case-insensitive; the key's letters are not.
            parts = re.split(r"(%[0-9A-Fa-f]{2})", encoded)
            patterns.add(
                "".join(
                    "(?i:" + re.escape(part) + ")"
                    if re.fullmatch(r"%[0-9A-Fa-f]{2}", part)
                    else re.escape(part)
                    for part in parts
                )
            )
        # One pass prevents a short key from matching the replacement marker.
        return re.sub("|".join(sorted(patterns, key=len, reverse=True)), "***REDACTED***", text)

    def _rate_limited_get(
        self,
        url: str,
        params: Optional[dict] = None,
        quiet: bool = False,
        *,
        _json_retry: bool = True,
    ) -> Optional[dict]:
        self._last_error = None
        if self.rate_limit_reached:
            self._last_error = "daily rate limit already reached"
            return None

        if params is None:
            params = {}

        elapsed = time.time() - self.last_call_time
        if elapsed < self.RATE_LIMIT_DELAY:
            time.sleep(self.RATE_LIMIT_DELAY - elapsed)

        try:
            response = self.session.get(url, params=params, timeout=self.timeout)
            self.last_call_time = time.time()
            self.api_calls_made += 1

            if response.status_code == 200:
                self.retry_count = 0
                try:
                    data = response.json()
                except ValueError as e:
                    # Malformed 200 body: retry once per call chain, then fail
                    # soft with a redacted reason instead of raising. The retry
                    # adds no backoff beyond the normal rate-limit delay and
                    # costs one extra API call (counts against any budget).
                    msg = self._redact_secret(f"invalid JSON response: {e}")
                    if _json_retry:
                        if not quiet:
                            print(f"WARNING: {msg}; retrying once", file=sys.stderr)
                        return self._rate_limited_get(url, params, quiet=quiet, _json_retry=False)
                    self._last_error = msg
                    if not quiet:
                        print(f"ERROR: {msg}", file=sys.stderr)
                    return None
                return data
            elif response.status_code == 429:
                self.retry_count += 1
                if self.retry_count <= self.max_retries:
                    print("WARNING: Rate limit exceeded. Waiting 60 seconds...", file=sys.stderr)
                    time.sleep(60)
                    return self._rate_limited_get(url, params, quiet=quiet, _json_retry=_json_retry)
                else:
                    self._last_error = "HTTP 429 (daily rate limit)"
                    print("ERROR: Daily API rate limit reached.", file=sys.stderr)
                    self.rate_limit_reached = True
                    return None
            else:
                body = self._redact_secret(response.text)[:200]
                msg = self._redact_secret(f"HTTP {response.status_code} - {body}")
                self._last_error = msg
                if not quiet:
                    print(
                        f"ERROR: API request failed: {msg}",
                        file=sys.stderr,
                    )
                return None
        except requests.exceptions.RequestException as e:
            redacted = self._redact_secret(str(e))
            self._last_error = f"request exception: {redacted}"
            print(f"ERROR: Request exception: {redacted}", file=sys.stderr)
            return None

    def _request_with_fallback(self, endpoint_key, symbols_str, extra_params=None):
        """Try stable endpoint first, fall back to v3 for legacy users.

        Returns parsed JSON in v3-compatible shape, or None if all fail.
        Non-last endpoints are called with quiet=True so the user isn't
        alarmed by an expected stable failure when v3 will catch it — but
        when a non-last endpoint DOES fail, a WARN line is emitted explaining
        why we're falling back. Otherwise users only see the (often misleading)
        last-endpoint error and have no clue what really went wrong.
        """
        params = dict(extra_params) if extra_params else {}
        endpoints = _FMP_ENDPOINTS[endpoint_key]
        is_single = "," not in symbols_str

        for i, (base_url, url_builder) in enumerate(endpoints):
            # Circuit breaker: skip endpoints with too many consecutive failures
            if base_url in self._disabled_endpoints:
                continue

            url, final_params = url_builder(base_url, symbols_str, dict(params))
            is_last = i == len(endpoints) - 1
            data = self._rate_limited_get(url, final_params, quiet=not is_last)
            if not data:  # falsy (None, [], {}) — try next endpoint
                self._record_endpoint_failure(base_url)
                self._warn_fallback(base_url, is_last, self._last_error)
                continue

            # Normalize new stable EOD flat-list shape to v3-compatible dict.
            # No-op for v3 dict / historicalStockList responses.
            # `timeseries` (original request) is passed as `limit` so the
            # EOD endpoint's full-history response is truncated to the
            # legacy "most-recent N rows" contract.
            if endpoint_key == "historical":
                limit = params.get("timeseries") if isinstance(params, dict) else None
                data = _normalize_eod_flat_list(data, symbols_str, limit=limit)
                if not data:
                    self._record_endpoint_failure(base_url)
                    self._warn_fallback(
                        base_url,
                        is_last,
                        f"response had no rows matching '{symbols_str}'",
                    )
                    continue

            # Shape validation: reject truthy-but-wrong-shape responses
            valid = True
            shape_issue: Optional[str] = None
            if endpoint_key == "historical":
                if not isinstance(data, dict):
                    valid = False
                    shape_issue = "expected dict"
                elif "historicalStockList" in data:
                    # stable batch format -> v3 single format (exact match only)
                    norm = symbols_str.replace("-", ".")
                    found = None
                    for entry in data["historicalStockList"]:
                        if entry.get("symbol", "").replace("-", ".") == norm:
                            found = {
                                "symbol": entry.get("symbol"),
                                "historical": entry.get("historical", []),
                            }
                            break
                    if found:
                        self._endpoint_failures[base_url] = 0
                        return found
                    valid = False
                    shape_issue = f"'{symbols_str}' not in historicalStockList"
                elif "historical" not in data:
                    valid = False
                    shape_issue = "missing 'historical' key"
                elif is_single and data.get("symbol"):
                    if data["symbol"].replace("-", ".") != symbols_str.replace("-", "."):
                        valid = False
                        shape_issue = (
                            f"response symbol '{data['symbol']}' != requested '{symbols_str}'"
                        )

            if valid:
                self._endpoint_failures[base_url] = 0
                return data
            self._record_endpoint_failure(base_url)
            self._warn_fallback(base_url, is_last, shape_issue or "unexpected response shape")
        return None

    def _warn_fallback(self, base_url: str, is_last: bool, reason: Optional[str]) -> None:
        """Emit a WARN line so users see why a non-last endpoint failed and the
        client is falling back. No-op when the failing endpoint is the last one
        (its error was already printed by _rate_limited_get with quiet=False)."""
        if is_last or not reason:
            return
        print(
            f"WARN: {base_url} failed ({reason}); falling back to next endpoint",
            file=sys.stderr,
        )

    def _record_endpoint_failure(self, base_url: str) -> None:
        """Track consecutive failures and disable endpoint after threshold."""
        failures = self._endpoint_failures.get(base_url, 0) + 1
        self._endpoint_failures[base_url] = failures
        if failures >= self._ENDPOINT_FAILURE_THRESHOLD:
            self._disabled_endpoints.add(base_url)

    def _request(self, url: str, params: dict, quiet: bool = False):
        """Compatibility entry point; HTTP/retry/redaction stays in the shared core."""
        return self._rate_limited_get(url, params=params, quiet=quiet)

    def _get(self, endpoint: str, params: Optional[dict] = None):
        """Route the dividend callers' existing names to supported stable endpoints."""
        if self.rate_limit_reached:
            return None
        query = dict(params or {})
        is_dividend = endpoint.startswith("historical-price-full/stock_dividend/")
        if is_dividend:
            name = "dividends"
            query["symbol"] = endpoint.rsplit("/", 1)[-1]
        elif endpoint == "stock-screener":
            name = "company-screener"
        else:
            name, separator, symbol = endpoint.partition("/")
            if separator:
                query["symbol"] = symbol
        allowed = {
            "company-screener",
            "income-statement",
            "balance-sheet-statement",
            "cash-flow-statement",
            "key-metrics",
            "ratios",
            "profile",
            "quote",
            "dividends",
        }
        if name not in allowed:
            # Endpoint strings are internal, but never interpolate an unknown
            # caller value into the URL or error text.
            print("ERROR: Unsupported dividend client endpoint", file=sys.stderr)
            return None
        data = self._request(f"{self.STABLE_URL}/{name}", query)
        if data is None:
            return None
        return self._normalize(endpoint, data)

    @staticmethod
    def _normalize(endpoint: str, data):
        """Keep legacy reader aliases without changing provider scale or sign."""
        if data is None:
            return None
        if endpoint.startswith("historical-price-full/stock_dividend/"):
            return {"historical": data} if isinstance(data, list) else data
        name = endpoint.partition("/")[0]
        if isinstance(data, list):
            for record in data:
                if not isinstance(record, dict):
                    continue
                if name == "key-metrics" and "roe" not in record and "returnOnEquity" in record:
                    record["roe"] = record["returnOnEquity"]
                if (
                    name == "cash-flow-statement"
                    and "dividendsPaid" not in record
                    and "netDividendsPaid" in record
                ):
                    record["dividendsPaid"] = record["netDividendsPaid"]
        return data

    def get_dividend_prices(self, symbol: str, days: int = 30) -> Optional[list[dict]]:
        """Most-recent-first history for the dividend adapters; no retired v3 retry."""
        url = f"{self.STABLE_URL}/historical-price-eod/full"
        if url in self._disabled_endpoints:
            return None
        today = _et_today()
        data = self._request(
            url,
            {
                "symbol": symbol,
                "from": (today - timedelta(days=days * 2 + 10)).isoformat(),
                "to": today.isoformat(),
            },
        )
        if isinstance(data, list) and data:
            self._endpoint_failures[url] = 0
            return data[:days]
        self._record_endpoint_failure(url)
        return None

    def get_historical_prices(self, symbol: str, days: int = 30) -> Optional[list[dict]]:
        """Fetch historical daily OHLCV data for a symbol.

        Args:
            symbol: Ticker symbol
            days: Number of trading days to fetch (default: 30)

        Returns:
            List of price dicts (most-recent-first) or None on failure.
        """
        cache_key = f"prices_{symbol}_{days}"
        if cache_key in self.cache:
            return self.cache[cache_key]

        data = self._request_with_fallback("historical", symbol, {"timeseries": days})
        if data and "historical" in data:
            result = data["historical"]
            self.cache[cache_key] = result
            return result
        return None

    def get_api_stats(self) -> dict:
        """Return API usage statistics."""
        return {
            "cache_entries": len(self.cache),
            "api_calls_made": self.api_calls_made,
            "rate_limit_reached": self.rate_limit_reached,
        }

"""WS-2: sector-aware payout-safety triad (GAAP / Adjusted / FCF).

Pure, dependency-free, offline-testable. Closes defect D2 (MKC payout
understated because a single GAAP-TTM proxy was inflated by a non-cash
acquisition remeasurement gain) and the over-strict utility -FCF problem.

Key rules (improvement-plan v2.1 R-3 + 4th-review):
- FCF payout is meaningless for banks and structurally negative for
  regulated utilities -> sector dispatch, NOT one uniform triad.
- Adjusted EPS is not in FMP; `adjusted_eps_source` is recorded and
  UNAVAILABLE -> verdict capped HOLD-REVIEW (fail-safe), never silent PASS.
- A merger completed within COMPLETED_MNA_LOOKBACK_QUARTERS presumes GAAP
  EPS is distorted -> force the adjusted path or HOLD-REVIEW
  (FITB/Comerica Q1-2026 GAAP EPS $0.15 is the golden case).
- Emits machine-checkable pre_order_blocker codes for WS-6 (4th-review #6).
"""

from __future__ import annotations

from dataclasses import dataclass, field

from thresholds import (
    ADJ_EPS_PAYOUT_CAUTION,
    ADJ_EPS_PAYOUT_MAX,
    BANK_DEPOSIT_BETA_HIGH,
    FCF_PAYOUT_HIGH_RISK,
    FCF_PAYOUT_MAX,
    GAAP_ADJ_DIVERGENCE,
    INSURER_OP_EPS_PAYOUT_MAX,
    UTILITY_FFO_DEBT_MIN,
)

_CONSUMER_SECTORS = {
    "consumer",
    "consumer defensive",
    "consumer staples",
    "consumer cyclical",
    "industrials",
    "industrial",
    "communication",
    "communication services",
    "technology",
    "healthcare",
    "energy",
    "materials",
    "basic materials",
    "real estate",  # treated as consumer-style triad unless flagged a REIT upstream
}
_BANK_SECTORS = {"bank", "banks", "financial", "financials", "financial services"}
_UTILITY_SECTORS = {"utility", "utilities"}
_INSURER_SECTORS = {"insurance", "insurer", "insurers"}


def _sector_kind(sector: str | None) -> str:
    s = (sector or "").strip().lower()
    if s in _BANK_SECTORS:
        return "bank"
    if s in _UTILITY_SECTORS:
        return "utility"
    if s in _INSURER_SECTORS:
        return "insurer"
    return "consumer"


def _ratio(numerator: float | None, denominator: float | None) -> float | None:
    if numerator is None or denominator is None or denominator == 0:
        return None
    return numerator / denominator


@dataclass
class SafetyAssessment:
    sector_kind: str
    safety_verdict: str  # PASS | CAUTION | FAIL | HOLD-REVIEW
    gaap_eps_payout: float | None = None
    adjusted_eps_payout: float | None = None
    fcf_payout: float | None = None
    adjusted_eps_source: str = "UNAVAILABLE"  # FMP | MANUAL | UNAVAILABLE
    gaap_adj_divergence: float | None = None
    one_off_flag: bool = False
    reasons: list[str] = field(default_factory=list)
    blockers: list[str] = field(default_factory=list)


def assess_payout_safety(
    *,
    sector: str | None,
    annual_dividend: float | None,
    gaap_eps: float | None = None,
    adjusted_eps: float | None = None,
    adjusted_eps_source: str = "UNAVAILABLE",
    fcf_per_share: float | None = None,
    completed_merger_within_4q: bool = False,
    bank_metrics: dict | None = None,
    utility_metrics: dict | None = None,
    insurer_metrics: dict | None = None,
) -> SafetyAssessment:
    kind = _sector_kind(sector)
    reasons: list[str] = []
    blockers: list[str] = []

    gaap_payout = _ratio(annual_dividend, gaap_eps)
    adj_payout = _ratio(annual_dividend, adjusted_eps)
    fcf_payout = _ratio(annual_dividend, fcf_per_share)

    # --- GAAP vs Adjusted divergence -> one-off flag (D2) ---
    one_off = False
    divergence = None
    if gaap_eps is not None and adjusted_eps not in (None, 0):
        divergence = abs(gaap_eps - adjusted_eps) / abs(adjusted_eps)
        if divergence > GAAP_ADJ_DIVERGENCE:
            one_off = True
            reasons.append(f"gaap_adj_divergence_{round(divergence * 100)}pct")
            blockers.append("gaap_adjusted_divergence_gt_25pct")

    # --- completed-merger linkage: GAAP presumed distorted (FITB/Comerica) ---
    if completed_merger_within_4q:
        reasons.append("completed_merger_within_4q_gaap_distorted")
        if adjusted_eps is None:
            blockers.append("adjusted_eps_unavailable")
            return SafetyAssessment(
                sector_kind=kind,
                safety_verdict="HOLD-REVIEW",
                gaap_eps_payout=_round(gaap_payout),
                adjusted_eps_payout=None,
                fcf_payout=_round(fcf_payout),
                adjusted_eps_source=adjusted_eps_source,
                gaap_adj_divergence=_round(divergence, 4),
                one_off_flag=True,
                reasons=reasons + ["use_adjusted_eps_after_merger"],
                blockers=blockers,
            )
        one_off = True

    # Adjusted EPS is the safety ANCHOR only for the consumer path. For
    # banks/utilities/insurers the sector module is the anchor, so a missing
    # adjusted EPS must NOT force HOLD-REVIEW there (5th-review F2). The
    # completed-merger GAAP-distortion path above still HOLDs regardless.
    adj_missing = adjusted_eps is None or adjusted_eps_source == "UNAVAILABLE"

    if kind == "bank":
        verdict = _assess_bank(bank_metrics or {}, gaap_payout, reasons, blockers)
    elif kind == "utility":
        verdict = _assess_utility(utility_metrics or {}, gaap_payout, reasons, blockers)
    elif kind == "insurer":
        verdict = _assess_insurer(insurer_metrics or {}, gaap_payout, adj_payout, reasons, blockers)
    else:
        if adj_missing:
            blockers.append("adjusted_eps_unavailable")
            reasons.append("adjusted_eps_unavailable_failsafe_hold")
        verdict = _assess_consumer(adj_payout, fcf_payout, adjusted_eps, reasons, blockers)
        if "adjusted_eps_unavailable" in blockers and verdict in ("PASS", "CAUTION"):
            verdict = "HOLD-REVIEW"

    return SafetyAssessment(
        sector_kind=kind,
        safety_verdict=verdict,
        gaap_eps_payout=_round(gaap_payout),
        adjusted_eps_payout=_round(adj_payout),
        fcf_payout=_round(fcf_payout),
        adjusted_eps_source=adjusted_eps_source,
        gaap_adj_divergence=_round(divergence, 4),
        one_off_flag=one_off,
        reasons=reasons,
        blockers=sorted(set(blockers)),
    )


def _round(value: float | None, ndigits: int = 4) -> float | None:
    return round(value, ndigits) if value is not None else None


def _assess_consumer(adj_payout, fcf_payout, adjusted_eps, reasons, blockers) -> str:
    # Primary: adjusted-EPS payout + FCF payout.
    if fcf_payout is not None and fcf_payout > FCF_PAYOUT_HIGH_RISK:
        reasons.append("fcf_payout_above_100pct")
        return "FAIL"
    if adj_payout is not None and adj_payout > ADJ_EPS_PAYOUT_MAX:
        reasons.append("adj_eps_payout_above_85pct")
        return "FAIL"
    caution = False
    if adj_payout is not None and adj_payout > ADJ_EPS_PAYOUT_CAUTION:
        reasons.append("adj_eps_payout_caution_band")
        caution = True
    if fcf_payout is not None and fcf_payout > FCF_PAYOUT_MAX:
        reasons.append("fcf_payout_above_80pct")
        caution = True
    if adjusted_eps is None:
        return "HOLD-REVIEW"
    return "CAUTION" if caution else "PASS"


def _assess_bank(m: dict, eps_payout, reasons, blockers) -> str:
    # FCF payout is meaningless for banks; use EPS payout + credit + capital.
    npl_trend = str(m.get("npl_trend", "")).lower()
    nco_trend = str(m.get("nco_trend", "")).lower()
    cet1 = m.get("cet1")
    cre_concentration = m.get("cre_concentration")
    verdict = "PASS"
    if cet1 is None:
        blockers.append("bank_capital_unavailable")
        verdict = "CAUTION"
    if cre_concentration is None:
        blockers.append("bank_cre_concentration_unavailable")
        verdict = "CAUTION"
    if npl_trend == "deteriorating" or nco_trend == "deteriorating":
        reasons.append("bank_npl_nco_deteriorating")
        blockers.append("bank_npl_nco_deteriorating")
        verdict = "CAUTION"
    deposit_beta = m.get("deposit_beta")
    db_str = str(deposit_beta).lower()
    if (isinstance(deposit_beta, (int, float)) and deposit_beta > BANK_DEPOSIT_BETA_HIGH) or (
        db_str in ("high", "rising")
    ):
        reasons.append("bank_deposit_beta_elevated")
        if verdict == "PASS":
            verdict = "CAUTION"
    if eps_payout is not None and eps_payout > ADJ_EPS_PAYOUT_MAX:
        reasons.append("bank_eps_payout_high")
        verdict = "FAIL"
    return verdict


def _assess_utility(m: dict, eps_payout, reasons, blockers) -> str:
    # Regulated utilities run structurally negative FCF (rate-base capex);
    # judge on FFO/debt + allowed ROE + rate-case + equity issuance instead.
    ffo_debt = m.get("ffo_to_debt")
    rate_case = str(m.get("rate_case_status", "")).lower()
    equity_issuance = str(m.get("equity_issuance_risk", "")).lower()
    allowed_roe_trend = str(m.get("allowed_roe_trend", "")).lower()
    verdict = "PASS"
    if ffo_debt is None:
        blockers.append("utility_ffo_debt_unavailable")
        verdict = "CAUTION"
    elif ffo_debt < UTILITY_FFO_DEBT_MIN:
        reasons.append(f"utility_ffo_debt_below_{UTILITY_FFO_DEBT_MIN}")
        blockers.append("utility_ffo_debt_weak")
        verdict = "CAUTION"
    if allowed_roe_trend in ("falling", "adverse"):
        reasons.append("utility_allowed_roe_falling")
        if verdict == "PASS":
            verdict = "CAUTION"
    if rate_case in ("adverse", "pending_adverse"):
        reasons.append("utility_rate_case_adverse")
        blockers.append("utility_rate_case_adverse")
        verdict = "HOLD-REVIEW"
    if equity_issuance in ("high", "dilutive"):
        reasons.append("utility_equity_issuance_risk")
        if verdict == "PASS":
            verdict = "CAUTION"
    if eps_payout is not None and eps_payout > 1.0:
        reasons.append("utility_eps_payout_above_100pct")
        verdict = "FAIL"
    return verdict


def _assess_insurer(m: dict, gaap_payout, op_eps_payout, reasons, blockers) -> str:
    combined_ratio = m.get("combined_ratio")
    reserve_dev = str(m.get("reserve_development", "")).lower()
    statutory = str(m.get("statutory_capital", "")).lower()
    verdict = "PASS"
    # Operating-EPS payout is the insurer income anchor (noisy GAAP).
    op_payout = op_eps_payout if op_eps_payout is not None else gaap_payout
    if op_payout is not None and op_payout > INSURER_OP_EPS_PAYOUT_MAX:
        reasons.append("insurer_operating_eps_payout_high")
        verdict = "FAIL"
    if combined_ratio is None:
        blockers.append("insurer_combined_ratio_unavailable")
        if verdict == "PASS":
            verdict = "CAUTION"
    if reserve_dev == "adverse":
        reasons.append("insurer_reserve_development_adverse")
        blockers.append("insurer_reserve_development_adverse")
        if verdict != "FAIL":  # never de-escalate a FAIL
            verdict = "HOLD-REVIEW"
    if combined_ratio is not None and combined_ratio > 1.0:
        reasons.append("insurer_combined_ratio_above_100")
        if verdict == "PASS":
            verdict = "CAUTION"
    if statutory == "weak":
        reasons.append("insurer_statutory_capital_weak")
        if verdict != "FAIL":
            verdict = "HOLD-REVIEW"
    return verdict

"""Shared fixtures for Kanchi dividend SOP script tests."""

import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

"""Tests for build_entry_signals.py."""

from build_entry_signals import (
    build_entry_row,
    load_tickers,
    normalize_metrics_yields,
    parse_ticker_csv,
)


def test_parse_ticker_csv_normalizes_and_deduplicates() -> None:
    tickers = parse_ticker_csv("aapl, MSFT, aapl, ko")
    assert tickers == ["AAPL", "MSFT", "KO"]


def test_load_tickers_from_json_candidates(tmp_path) -> None:
    path = tmp_path / "input.json"
    path.write_text('{"candidates":[{"ticker":"jnj"},{"ticker":"pg"}]}', encoding="utf-8")
    assert load_tickers(path, None) == ["JNJ", "PG"]


def test_load_tickers_from_screener_stocks_symbol_contract(tmp_path) -> None:
    """Both optional upstream screeners emit ``stocks[].symbol``."""
    path = tmp_path / "screener-output.json"
    path.write_text(
        '{"stocks":[{"symbol":"jnj"},{"symbol":"PG"},{"symbol":"jnj"}]}',
        encoding="utf-8",
    )
    assert load_tickers(path, None) == ["JNJ", "PG"]


def test_normalize_metrics_yields_limits_to_five_points() -> None:
    metrics = [
        {"dividendYield": 0.02},
        {"dividendYield": 0.03},
        {"dividendYield": 0.04},
        {"dividendYield": 0.05},
        {"dividendYield": 0.06},
        {"dividendYield": 0.07},
    ]
    assert normalize_metrics_yields(metrics) == [2.0, 3.0, 4.0, 5.0, 6.0]


def test_build_entry_row_wait_signal() -> None:
    row = build_entry_row(
        ticker="AAPL",
        alpha_pp=0.5,
        quote={"price": 200.0},
        profile={"lastDiv": 4.0},
        key_metrics=[
            {"dividendYield": 0.02},
            {"dividendYield": 0.021},
            {"dividendYield": 0.019},
            {"dividendYield": 0.022},
            {"dividendYield": 0.018},
        ],
    )
    assert row["signal"] == "WAIT"
    assert row["target_yield_pct"] == 2.5
    assert row["buy_target_price"] == 160.0
    assert row["drop_needed_pct"] == 20.0


def test_build_entry_row_triggered_signal() -> None:
    row = build_entry_row(
        ticker="KO",
        alpha_pp=0.5,
        quote={"price": 45.0},
        profile={"lastDiv": 2.4},
        key_metrics=[
            {"dividendYield": 0.03},
            {"dividendYield": 0.032},
            {"dividendYield": 0.031},
            {"dividendYield": 0.033},
            {"dividendYield": 0.034},
        ],
    )
    assert row["signal"] == "TRIGGERED"
    assert row["buy_target_price"] == 64.86
    assert row["drop_needed_pct"] == 0.0


def test_build_entry_row_assumption_required_when_missing_data() -> None:
    row = build_entry_row(
        ticker="XXX",
        alpha_pp=0.5,
        quote=None,
        profile=None,
        key_metrics=[],
    )
    assert row["signal"] == "ASSUMPTION-REQUIRED"
    assert "quote_missing" in row["notes"]
    assert "profile_missing" in row["notes"]


def test_build_entry_row_attaches_ws2_payout_safety_and_blockers() -> None:
    # WS-2 integration: financials -> payout_safety + pre_order_blockers.
    row = build_entry_row(
        ticker="MKC",
        alpha_pp=0.5,
        quote={"price": 46.35},
        profile={"lastDiv": 1.92, "sector": "Consumer Staples"},
        key_metrics=[{"dividendYield": 0.04}],
        financials={
            "sector": "Consumer Staples",
            "gaap_eps": 12.9,
            "adjusted_eps": 3.09,
            "adjusted_eps_source": "MANUAL",
        },
    )
    assert row["payout_safety"]["one_off_flag"] is True
    assert "gaap_one_off" in row["notes"]
    assert "gaap_adjusted_divergence_gt_25pct" in row["pre_order_blockers"]


def test_build_entry_row_applies_ws3_event_cap() -> None:
    from event_scanner import MAJOR_EVENT, ScanResult

    row = build_entry_row(
        ticker="MKC",
        alpha_pp=0.5,
        quote={"price": 46.0},
        profile={"lastDiv": 1.92},
        key_metrics=[{"dividendYield": 0.04}],
        event_scan=ScanResult(
            ticker="MKC", result=MAJOR_EVENT, pending_mna=True, reasons=["tx_value_280pct_mcap"]
        ),
    )
    assert row["verdict_cap"] == "HOLD-REVIEW"
    assert row["t1_blocked"] is True
    assert "major_structural_event" in row["pre_order_blockers"]
    assert row["event_scan"]["pending_mna"] is True


def test_build_entry_row_ws5_verdict_and_provenance() -> None:
    from datetime import date, timedelta

    # CFR D5: latest declared raise -> near-floor -> STEP1-RECHECK verdict.
    hist = [
        {
            "date": (date(2023, 5, 30) + timedelta(days=91 * i)).isoformat(),
            "dividend": (1.00 if i < 12 else 1.03),
            "label": "cash",
        }
        for i in range(13)
    ]
    row = build_entry_row(
        ticker="CFR",
        alpha_pp=0.5,
        quote={"price": 134.70},
        profile={"lastDiv": 4.12, "sector": "Financial Services"},
        key_metrics=[{"dividendYield": 0.031}],
        dividend_history=hist,
        floor_pct=3.0,
    )
    assert row["verdict"] == "STEP1-RECHECK"
    assert row["t1_blocked"] is True
    assert row["provenance"]["dividend_source"] == "fmp_stock_dividend"
    assert "evidence_refs" in row["provenance"]

"""Tests for build_sop_plan.py."""

import json
from pathlib import Path

from build_sop_plan import load_candidates, parse_ticker_csv, render_markdown


def test_parse_ticker_csv_normalizes_and_deduplicates() -> None:
    tickers = parse_ticker_csv("jnj, PG, jnj,  ko ")
    assert tickers == ["JNJ", "PG", "KO"]


def test_load_candidates_from_json_with_bucket(tmp_path: Path) -> None:
    payload = {
        "profile": "income_now",
        "candidates": [{"ticker": "jnj", "bucket": "core"}, {"ticker": "o", "bucket": "satellite"}],
    }
    input_path = tmp_path / "input.json"
    input_path.write_text(json.dumps(payload), encoding="utf-8")
    candidates, profile = load_candidates(input_path, None)
    assert profile == "income_now"
    assert candidates == [
        {"ticker": "JNJ", "bucket": "core"},
        {"ticker": "O", "bucket": "satellite"},
    ]


def test_load_candidates_from_screener_stocks_symbol_contract(tmp_path: Path) -> None:
    """Workflow screener artifacts use ``stocks[].symbol``, not candidates/ticker."""
    payload = {"stocks": [{"symbol": "jnj"}, {"symbol": "o"}]}
    input_path = tmp_path / "screener-output.json"
    input_path.write_text(json.dumps(payload), encoding="utf-8")

    candidates, profile = load_candidates(input_path, None)

    assert profile == "balanced"
    assert candidates == [
        {"ticker": "JNJ", "bucket": "unassigned"},
        {"ticker": "O", "bucket": "unassigned"},
    ]


def test_load_candidates_from_ticker_csv() -> None:
    candidates, profile = load_candidates(None, "aapl,msft")
    assert profile == "balanced"
    assert candidates == [
        {"ticker": "AAPL", "bucket": "unassigned"},
        {"ticker": "MSFT", "bucket": "unassigned"},
    ]


def test_render_markdown_contains_expected_sections() -> None:
    markdown = render_markdown(
        candidates=[{"ticker": "JNJ", "bucket": "core"}],
        as_of="2026-02-22",
        profile="balanced",
    )
    assert "# Kanchi SOP Plan" in markdown
    assert "## Candidate Universe" in markdown
    assert "| JNJ | core |" in markdown

"""WS-1 golden + edge-case tests for dividend_basis.py (offline, deterministic).

Fixtures model the FMP `/stable dividends` record shape (date / dividend /
declarationDate) and the real dividend cadences observed on 2026-05-17.
Verdicts frozen here are the P0 regression gate for defects D1 (freeze),
D4 (special/variable), D5 (stale-dividend false-negative near the floor).

# DATA-DATE: 2026-05-17
# VALID-UNTIL: 2026-08-31 (re-confirm CMCSA freeze / CFR raise next quarter)
# REVIEW-TRIGGER: CMCSA dividend restart, CFR dividend change
"""

from datetime import date, timedelta

from dividend_basis import analyze_dividends, step1_decision


def _series(start: date, n: int, amounts, step_days: int = 91, label="cash"):
    """Build a newest-last dividend history; amounts is a value or list."""
    out = []
    for i in range(n):
        amt = amounts[i] if isinstance(amounts, (list, tuple)) else amounts
        out.append(
            {
                "date": (start + timedelta(days=step_days * i)).isoformat(),
                "dividend": amt,
                "label": label,
            }
        )
    return out


# --- D4: CALM variable-dividend policy -> FAIL ---
def test_calm_variable_policy_flag_and_fail():
    hist = _series(
        date(2023, 2, 1),
        12,
        [0.006, 0.116, 0.755, 0.77, 0.997, 1.019, 1.378, 1.489, 3.495, 2.354, 0.723, 0.357],
    )
    b = analyze_dividends(hist, price=76.88, floor_pct=4.0)
    assert b.variable_policy_flag is True
    verdict, _ = step1_decision(b, 4.0)
    assert verdict == "FAIL"


# --- D4 + v2.1 R-1: ORI specials removed BEFORE variable test ---
def test_ori_special_excluded_not_variable():
    # Steady growing regular quarterly + two large annual specials.
    hist = []
    reg = [0.245, 0.265, 0.265, 0.265, 0.265, 0.29, 0.29, 0.29, 0.315]
    d = date(2023, 12, 1)
    for i, amt in enumerate(reg):
        hist.append(
            {"date": (d + timedelta(days=91 * i)).isoformat(), "dividend": amt, "label": "cash"}
        )
    hist.append({"date": "2025-01-03", "dividend": 2.00, "label": "cash"})
    hist.append({"date": "2026-01-02", "dividend": 2.50, "label": "cash"})
    b = analyze_dividends(hist, price=39.32, floor_pct=3.0)
    assert b.special_dividend_flag is True  # specials detected by amount-outlier
    assert b.variable_policy_flag is False  # residual is steady (R-1 ordering)
    assert b.ttm_yield_pct > b.regular_forward_yield_pct  # specials inflate ttm only


# --- D1: freeze via time-series path ---
def test_freeze_timepath_holds_review():
    hist = _series(date(2022, 3, 1), 13, 0.33)  # >3y flat, no increase ever
    b = analyze_dividends(hist, price=24.76, floor_pct=4.0)
    assert b.freeze_flag is True
    assert b.cut_flag is False
    verdict, reason = step1_decision(b, 4.0)
    assert verdict == "HOLD-REVIEW" and reason == "dividend_freeze"


# --- D1: freeze via issuer language when YoY rate unavailable (short history) ---
def test_freeze_language_path():
    hist = _series(date(2025, 6, 1), 4, 0.33)  # only 4 pays -> no prior-year rate
    b = analyze_dividends(
        hist,
        price=24.76,
        floor_pct=4.0,
        issuer_language="Comcast maintains dividend unchanged for 2026",
    )
    assert b.freeze_flag is True


# --- D5: CFR near-floor on the LATEST declared raise -> RECHECK, not FAIL ---
def test_cfr_d5_freshness_recheck_not_fail():
    hist = _series(date(2023, 5, 30), 13, [1.00] * 12 + [1.03])  # latest declared raise to 1.03
    b = analyze_dividends(hist, price=134.70, floor_pct=3.0)
    # 1.03 * 4 / 134.70 = 3.06% -> just above 3.0 floor, inside freshness band
    assert b.latest_declared_annualized == 4.12
    assert b.floor_borderline is True
    verdict, _ = step1_decision(b, 3.0, source_confirmed=False)
    assert verdict == "STEP1-RECHECK"  # NOT "FAIL" (the D5 bug)
    verdict_ok, _ = step1_decision(b, 3.0, source_confirmed=True)
    assert verdict_ok in ("STEP1-PASS", "HOLD-REVIEW")


# --- cut detected and FAIL'd, distinct from freeze ---
def test_cut_flag_distinct_from_freeze():
    hist = _series(date(2023, 1, 1), 12, [0.50] * 8 + [0.40] * 4)  # recent cut
    b = analyze_dividends(hist, price=30.0, floor_pct=3.0)
    assert b.cut_flag is True
    assert b.freeze_flag is False
    assert step1_decision(b, 3.0)[0] == "FAIL"


# --- monthly REIT must NOT misfire variable_policy_flag ---
def test_monthly_reit_no_variable_misfire():
    hist = _series(date(2024, 1, 15), 24, 0.2625, step_days=30, label="cash")
    b = analyze_dividends(hist, price=58.0, floor_pct=4.0)
    assert b.cadence == "monthly" and b.pays_per_year == 12
    assert b.variable_policy_flag is False
    assert b.cut_flag is False and b.freeze_flag is True  # flat & old -> freeze


# --- insufficient history -> ASSUMPTION-REQUIRED, not a hard verdict ---
def test_annual_only_short_history_assumption_required():
    hist = _series(date(2024, 6, 1), 2, [1.0, 1.05], step_days=365)
    b = analyze_dividends(hist, price=50.0, floor_pct=3.0)
    assert b.status == "ASSUMPTION-REQUIRED"
    assert step1_decision(b, 3.0)[0] == "STEP1-RECHECK"


def test_no_dividend_history():
    b = analyze_dividends([], price=10.0, floor_pct=3.0)
    assert b.status == "NO-DIVIDEND"
    assert step1_decision(b, 3.0)[0] == "FAIL"


# --- 4th-review point 10: weak payout language must NOT flag variable ---
def test_weak_payout_language_does_not_flag_variable():
    hist = _series(date(2023, 1, 1), 12, 0.50)  # steady regular
    b = analyze_dividends(
        hist,
        price=40.0,
        floor_pct=3.0,
        issuer_language="We target a dividend payout ratio of 40% of net income.",
    )
    assert b.variable_policy_flag is False
    assert any("weak_payout_language" in r for r in b.reasons)


def test_strong_variable_language_flags_variable():
    hist = _series(date(2023, 1, 1), 12, 0.50)
    b = analyze_dividends(
        hist,
        price=40.0,
        floor_pct=3.0,
        issuer_language="The company follows a variable dividend policy.",
    )
    assert b.variable_policy_flag is True


# --- 4th-review point 4: overdue declaration -> suspension_flag -> FAIL ---
def test_suspension_flag_when_declaration_overdue():
    hist = _series(date(2023, 1, 1), 12, 0.50)  # last pay ~2025-09
    b = analyze_dividends(hist, price=40.0, floor_pct=3.0, as_of_date="2026-05-17")
    assert b.suspension_flag is True
    assert step1_decision(b, 3.0)[0] == "FAIL"


def test_no_suspension_when_recent_and_dates_used_present():
    hist = _series(date(2025, 6, 1), 8, 0.50)
    b = analyze_dividends(hist, price=40.0, floor_pct=3.0, as_of_date="2026-05-17")
    assert b.suspension_flag is False
    assert len(b.dividend_dates_used) >= 4  # audit trail populated

"""WS-3 event-scanner tests (offline, NO live web — CR-2).

# DATA-DATE: 2026-05-17
# REVIEW-TRIGGER: MKC-Unilever close (mid-2027), WTRG-AWK close
"""

from event_scanner import (
    CLEAN_CONFIRMED,
    FAILED_DEGRADED,
    MAJOR_EVENT,
    MINOR_EVENT_CAUTION,
    NO_EVENT_FOUND,
    SKIPPED,
    FixtureEventScanner,
    ManualEventScanner,
    ScanResult,
    apply_event_cap,
    is_major_structural_event,
)


# --- D3: MKC-Unilever is a major structural event ---
def test_mkc_unilever_is_major_structural_event():
    major, reasons = is_major_structural_event(
        {
            "tx_value": 44_800,
            "market_cap": 16_000,
            "share_issuance_pct": 90,
            "structure": "merger_of_equals",
            "control_change": True,
        }
    )
    assert major is True
    assert any("tx_value" in r for r in reasons)


def test_small_bolt_on_is_not_major():
    major, reasons = is_major_structural_event(
        {"tx_value": 200, "market_cap": 30_000, "share_issuance_pct": 0}
    )
    assert major is False and reasons == []


def test_sector_specific_utility_materiality():
    major, reasons = is_major_structural_event(
        {
            "market_cap": 20_000,
            "tx_value": 0,
            "sector": "utilities",
            "sector_metrics": {
                "acq_or_capex_pct_rate_base": 12,
                "regulatory_approval_pending": True,
            },
        }
    )
    assert major is True
    assert "utility_acq_capex_gt_10pct_rate_base" in reasons


def test_rolling_24m_cumulative_mna():
    major, reasons = is_major_structural_event(
        {"market_cap": 10_000, "tx_value": 100, "rolling_24m_tx_value": 1_800}
    )
    assert major is True
    assert "rolling_24m_tx_gt_15pct_mcap" in reasons


# --- pessimistic cap (4th-review #5) ---
def test_major_event_caps_and_blocks_t1():
    scan = ScanResult(ticker="MKC", result=MAJOR_EVENT, reasons=["tx_value_280pct_mcap"])
    cap = apply_event_cap(scan, step5_triggered=False)
    assert cap["verdict_cap"] == "HOLD-REVIEW"
    assert cap["t1_blocked"] is True
    assert "major_structural_event" in cap["blockers"]


def test_skipped_scan_on_triggered_name_caps_hold_review():
    scan = ScanResult(ticker="CMCSA", result=SKIPPED)
    cap = apply_event_cap(scan, step5_triggered=True)
    assert cap["verdict_cap"] == "HOLD-REVIEW"
    assert cap["t1_blocked"] is True
    assert "event_scan_failed_or_skipped" in cap["blockers"]


def test_failed_scan_on_non_triggered_blocks_t1_only():
    scan = ScanResult(ticker="XYZ", result=FAILED_DEGRADED)
    cap = apply_event_cap(scan, step5_triggered=False)
    assert cap["verdict_cap"] is None  # not entry-ready -> no hard verdict cap
    assert cap["t1_blocked"] is True


def test_no_event_found_is_weaker_than_clean_confirmed():
    weak = apply_event_cap(ScanResult("A", NO_EVENT_FOUND), step5_triggered=True)
    strong = apply_event_cap(ScanResult("B", CLEAN_CONFIRMED), step5_triggered=True)
    assert weak["verdict_cap"] == "HOLD-REVIEW"
    assert strong["verdict_cap"] is None and strong["t1_blocked"] is False


def test_minor_event_is_caution_note_only():
    cap = apply_event_cap(ScanResult("C", MINOR_EVENT_CAUTION), step5_triggered=True)
    assert cap["verdict_cap"] is None and cap["t1_blocked"] is False


# --- scanners are injectable; unknown ticker is pessimistic NO_EVENT_FOUND ---
def test_fixture_scanner_unknown_ticker_pessimistic():
    s = FixtureEventScanner({"MKC": ScanResult("MKC", MAJOR_EVENT)})
    assert s.scan("MKC", "2026-05-17").result == MAJOR_EVENT
    assert s.scan("ZZZ", "2026-05-17").result == NO_EVENT_FOUND


def test_manual_scanner_missing_file_is_pessimistic(tmp_path):
    s = ManualEventScanner(tmp_path / "absent.json")
    assert s.scan("MKC", "2026-05-17").result == NO_EVENT_FOUND


def test_manual_scanner_reads_events(tmp_path):
    p = tmp_path / "events.json"
    p.write_text(
        '{"events": {"MKC": {"result": "MAJOR_EVENT", "pending_mna": true}}}', encoding="utf-8"
    )
    s = ManualEventScanner(p)
    r = s.scan("mkc", "2026-05-17")
    assert r.result == MAJOR_EVENT and r.pending_mna is True

"""5th-review integration-gap fixes F1-F4 (offline, deterministic)."""

from datetime import date, timedelta

from build_entry_signals import build_entry_row
from dividend_basis import analyze_dividends, step1_decision
from payout_safety import assess_payout_safety


def _q(start, n, amt, step=91, decl=False):
    out = []
    for i in range(n):
        rec = {
            "date": (start + timedelta(days=step * i)).isoformat(),
            "dividend": (amt[i] if isinstance(amt, list) else amt),
            "label": "cash",
        }
        if decl:
            rec["declarationDate"] = (start + timedelta(days=step * i - 30)).isoformat()
        out.append(rec)
    return out


# --- F4: confirmed-source path via declarationDate ---
def test_f4_confirmed_declaration_resolves_near_floor():
    hist = _q(date(2023, 5, 30), 13, [1.00] * 12 + [1.03], decl=True)
    b = analyze_dividends(hist, price=134.70, floor_pct=3.0)
    assert b.latest_declared_confirmed is True
    # 3.06% > 3.0 floor, borderline, but board-declared -> resolves (not stuck)
    v, _ = step1_decision(b, 3.0, source_confirmed=b.latest_declared_confirmed)
    assert v == "STEP1-PASS"


def test_f4_unconfirmed_stays_recheck():
    hist = _q(date(2023, 5, 30), 13, [1.00] * 12 + [1.03], decl=False)
    b = analyze_dividends(hist, price=134.70, floor_pct=3.0)
    assert b.latest_declared_confirmed is False
    v, _ = step1_decision(b, 3.0, source_confirmed=b.latest_declared_confirmed)
    assert v == "STEP1-RECHECK"


def test_f4_cli_path_cfr_confirmed_not_recheck():
    hist = _q(date(2023, 5, 30), 13, [1.00] * 12 + [1.03], decl=True)
    row = build_entry_row(
        ticker="CFR",
        alpha_pp=0.5,
        quote={"price": 134.70},
        profile={"sector": "Financial Services"},
        key_metrics=[{"dividendYield": 0.031}],
        dividend_history=hist,
        floor_pct=3.0,
    )
    assert row["step1_verdict"] == "STEP1-PASS"  # confirmed declaration


# --- F3: as_of threaded so suspension is reachable from build_entry_row ---
def test_f3_as_of_makes_suspension_reachable():
    hist = _q(date(2023, 1, 1), 12, 0.50)  # last pay ~2025-09
    row = build_entry_row(
        ticker="SUSP",
        alpha_pp=0.5,
        quote={"price": 40.0},
        profile={"sector": "Industrials"},
        key_metrics=[{"dividendYield": 0.05}],
        dividend_history=hist,
        floor_pct=3.0,
        as_of="2026-05-17",
    )
    assert row["dividend_basis"]["suspension_flag"] is True
    assert row["verdict"] == "FAIL"


# --- F2: adjusted-EPS unavailable must NOT HOLD a bank (sector is anchor) ---
def test_f2_bank_adjusted_unavailable_not_hold_review():
    a = assess_payout_safety(
        sector="financial services",
        annual_dividend=1.88,
        gaap_eps=6.17,
        adjusted_eps=None,
        adjusted_eps_source="UNAVAILABLE",
        bank_metrics={
            "cet1": 0.11,
            "cre_concentration": "low",
            "npl_trend": "stable",
            "nco_trend": "stable",
        },
    )
    assert "adjusted_eps_unavailable" not in a.blockers
    assert a.safety_verdict == "PASS"  # sector anchor governs, not adj EPS


def test_f2_consumer_adjusted_unavailable_still_holds():
    a = assess_payout_safety(
        sector="consumer staples",
        annual_dividend=1.0,
        gaap_eps=3.0,
        adjusted_eps=None,
        adjusted_eps_source="UNAVAILABLE",
        fcf_per_share=2.5,
    )
    assert "adjusted_eps_unavailable" in a.blockers
    assert a.safety_verdict == "HOLD-REVIEW"


def test_f2_bank_deposit_beta_elevated_caution():
    a = assess_payout_safety(
        sector="banks",
        annual_dividend=1.0,
        gaap_eps=5.0,
        adjusted_eps=5.0,
        adjusted_eps_source="FMP",
        bank_metrics={
            "cet1": 0.12,
            "cre_concentration": "low",
            "npl_trend": "stable",
            "nco_trend": "stable",
            "deposit_beta": 0.7,
        },
    )
    assert a.safety_verdict == "CAUTION"
    assert "bank_deposit_beta_elevated" in a.reasons


def test_f2_utility_ffo_debt_below_min_caution():
    a = assess_payout_safety(
        sector="utilities",
        annual_dividend=1.5,
        gaap_eps=2.5,
        adjusted_eps=2.5,
        adjusted_eps_source="FMP",
        utility_metrics={"ffo_to_debt": 0.10, "rate_case_status": "constructive"},
    )
    assert "utility_ffo_debt_weak" in a.blockers
    assert a.safety_verdict == "CAUTION"


def test_f2_utility_allowed_roe_falling_caution():
    a = assess_payout_safety(
        sector="utilities",
        annual_dividend=1.5,
        gaap_eps=2.5,
        adjusted_eps=2.5,
        adjusted_eps_source="FMP",
        utility_metrics={
            "ffo_to_debt": 0.18,
            "rate_case_status": "constructive",
            "allowed_roe_trend": "falling",
        },
    )
    assert "utility_allowed_roe_falling" in a.reasons
    assert a.safety_verdict == "CAUTION"


def test_f2_insurer_operating_eps_payout_high_fails():
    a = assess_payout_safety(
        sector="insurance",
        annual_dividend=4.0,
        gaap_eps=4.0,
        adjusted_eps=4.0,
        adjusted_eps_source="FMP",
        insurer_metrics={
            "combined_ratio": 0.95,
            "reserve_development": "favorable",
            "statutory_capital": "strong",
        },
    )
    # op-EPS payout = 4.0/4.0 = 100% > 85% ceiling -> FAIL, not downgraded
    assert a.safety_verdict == "FAIL"

"""WS-7b: expanded sector-dispatch golden gate (end-to-end, offline).

Locks WS-2/WS-4 sector behaviour through build_entry_row so a future
threshold/dispatch edit cannot silently regress the bank / utility /
insurer paths that the 2026-05 runs got wrong.

# DATA-DATE: 2026-05-17
# VALID-UNTIL: 2026-08-31
# REVIEW-TRIGGER: OZK credit-trend reversal, WTRG/AWK close, ORI special cadence
"""

from datetime import date, timedelta

from build_entry_signals import build_entry_row
from event_scanner import CLEAN_CONFIRMED, ScanResult

# Step 4b ran clean — required for a PASS tier on a TRIGGERED name.
_CLEAN = ScanResult(ticker="_", result=CLEAN_CONFIRMED)


def _q(start, n, amt, step=91):
    amounts = amt if isinstance(amt, list) else [amt] * n
    return [
        {
            "date": (start + timedelta(days=step * i)).isoformat(),
            "dividend": amounts[i],
            "label": "cash",
        }
        for i in range(n)
    ]


_RAISER = [0.40, 0.41, 0.42, 0.43, 0.43, 0.44, 0.45, 0.46, 0.46, 0.47, 0.48, 0.49, 0.50]


# OZK-style: strong serial raiser but deteriorating credit -> PASS-CAUTION.
def test_ws7b_bank_credit_deterioration_pass_caution():
    hist = _q(date(2023, 1, 1), 13, _RAISER)
    row = build_entry_row(
        ticker="OZK",
        alpha_pp=0.5,
        quote={"price": 46.73},
        profile={"sector": "Financial Services"},
        key_metrics=[{"dividendYield": 0.04}],
        dividend_history=hist,
        floor_pct=3.0,
        financials={
            "sector": "Financial Services",
            "gaap_eps": 6.17,
            "adjusted_eps": 6.17,
            "adjusted_eps_source": "FMP",
            "fcf_per_share": -50.0,  # must be ignored for a bank
            "bank_metrics": {
                "cet1": 0.11,
                "cre_concentration": "high",
                "npl_trend": "deteriorating",
                "nco_trend": "deteriorating",
            },
        },
        event_scan=_CLEAN,
    )
    assert row["payout_safety"]["sector_kind"] == "bank"
    assert row["verdict"] == "PASS-CAUTION"
    assert "bank_npl_nco_deteriorating" in row["pre_order_blockers"]


# EXC-style regulated utility: structurally negative FCF must NOT auto-FAIL.
def test_ws7b_utility_negative_fcf_not_fail():
    hist = _q(date(2023, 1, 1), 13, _RAISER)
    row = build_entry_row(
        ticker="EXC",
        alpha_pp=0.5,
        quote={"price": 43.38},
        profile={"sector": "Utilities"},
        key_metrics=[{"dividendYield": 0.037}],
        dividend_history=hist,
        floor_pct=3.0,
        financials={
            "sector": "Utilities",
            "gaap_eps": 2.74,
            "adjusted_eps": 2.74,
            "adjusted_eps_source": "FMP",
            "fcf_per_share": -3.0,
            "utility_metrics": {"ffo_to_debt": 0.16, "rate_case_status": "constructive"},
        },
        event_scan=_CLEAN,
    )
    assert row["payout_safety"]["sector_kind"] == "utility"
    assert row["verdict"] in ("CLEAN-PASS", "PASS-CAUTION")  # NOT FAIL


# ORI-style insurer: regular yield clears the 3% floor; specials excluded.
def test_ws7b_insurer_regular_yield_passes_floor():
    reg = _q(date(2023, 12, 1), 10, 0.30)
    reg.append({"date": "2026-01-02", "dividend": 2.50, "label": "cash"})  # special
    row = build_entry_row(
        ticker="ORI",
        alpha_pp=0.5,
        quote={"price": 37.0},
        profile={"sector": "Insurance"},
        key_metrics=[{"dividendYield": 0.032}],
        dividend_history=reg,
        floor_pct=3.0,
        financials={
            "sector": "Insurance",
            "gaap_eps": 3.7,
            "adjusted_eps": 3.7,
            "adjusted_eps_source": "FMP",
            "insurer_metrics": {
                "combined_ratio": 0.95,
                "reserve_development": "favorable",
                "statutory_capital": "strong",
            },
        },
        event_scan=_CLEAN,
    )
    db = row["dividend_basis"]
    assert db["special_dividend_flag"] is True
    assert db["variable_policy_flag"] is False
    assert row["verdict"] != "FAIL"  # not a trap; insurer-module review path

"""WS-7a: P0 GOLDEN REGRESSION GATE (end-to-end, offline, deterministic).

This module is the merge gate for P0 (CR-1 fix): it pins the end-to-end
build_entry_row verdict for every known-hard case that previously produced
a wrong call. A future SOP/threshold edit that regresses any of these MUST
fail here. No live FMP / WebSearch — all inputs are frozen fixtures.

# DATA-DATE: 2026-05-17
# VALID-UNTIL: 2026-08-31
# REVIEW-TRIGGER: CMCSA dividend restart, CFR dividend change,
#                 MKC-Unilever close (~mid-2027), FITB post-Comerica normalize
"""

from datetime import date, timedelta

from build_entry_signals import build_entry_row
from event_scanner import CLEAN_CONFIRMED, MAJOR_EVENT, ScanResult

# Step 4b executed and found nothing material. Required for any PASS tier on
# a TRIGGERED name (6th-review High2: a missing scan pessimistically caps).
_CLEAN = ScanResult(ticker="_", result=CLEAN_CONFIRMED)


def _q(start: date, n: int, amounts, step=91):
    return [
        {
            "date": (start + timedelta(days=step * i)).isoformat(),
            "dividend": (amounts[i] if isinstance(amounts, list) else amounts),
            "label": "cash",
        }
        for i in range(n)
    ]


def _verdict(**kw):
    return build_entry_row(alpha_pp=0.5, key_metrics=[{"dividendYield": 0.04}], **kw)


# --- CALM: variable dividend policy -> FAIL ---
def test_golden_calm_fail():
    hist = _q(
        date(2023, 2, 1),
        12,
        [0.006, 0.116, 0.755, 0.77, 0.997, 1.019, 1.378, 1.489, 3.495, 2.354, 0.723, 0.357],
    )
    row = _verdict(
        ticker="CALM",
        quote={"price": 76.88},
        profile={"sector": "Consumer Defensive"},
        dividend_history=hist,
        floor_pct=4.0,
    )
    assert row["verdict"] == "FAIL"


# --- ORI: specials removed before variable test; not a CALM-style payer ---
def test_golden_ori_special_not_variable():
    reg = _q(date(2023, 12, 1), 9, [0.245, 0.265, 0.265, 0.265, 0.265, 0.29, 0.29, 0.29, 0.315])
    reg += [
        {"date": "2025-01-03", "dividend": 2.00, "label": "cash"},
        {"date": "2026-01-02", "dividend": 2.50, "label": "cash"},
    ]
    row = _verdict(
        ticker="ORI",
        quote={"price": 39.32},
        profile={"sector": "Insurance"},
        dividend_history=reg,
        floor_pct=3.0,
    )
    db = row["dividend_basis"]
    assert db["special_dividend_flag"] is True
    assert db["variable_policy_flag"] is False
    assert row["verdict"] != "FAIL"  # not wrongly killed as variable


# --- CMCSA: 2026 freeze + strong safety -> CONDITIONAL-PASS (income cash-cow) ---
def test_golden_cmcsa_freeze_conditional_pass():
    hist = _q(date(2022, 3, 1), 13, 0.33)  # multi-year flat
    row = _verdict(
        ticker="CMCSA",
        quote={"price": 24.76},
        profile={"sector": "Communication Services"},
        dividend_history=hist,
        floor_pct=4.0,
        financials={
            "sector": "Communication Services",
            "gaap_eps": 4.2,
            "adjusted_eps": 4.2,
            "adjusted_eps_source": "FMP",
            "fcf_per_share": 4.5,
        },
        event_scan=_CLEAN,
    )
    assert row["dividend_basis"]["freeze_flag"] is True
    assert row["verdict"] == "CONDITIONAL-PASS"


# --- MKC: pending mega-merger -> HOLD-REVIEW + T1 blocked ---
def test_golden_mkc_event_hold_review():
    hist = _q(date(2023, 1, 1), 13, 0.45)
    row = _verdict(
        ticker="MKC",
        quote={"price": 46.35},
        profile={"sector": "Consumer Defensive"},
        dividend_history=hist,
        floor_pct=3.0,
        event_scan=ScanResult(
            ticker="MKC", result=MAJOR_EVENT, pending_mna=True, reasons=["tx_value_280pct_mcap"]
        ),
    )
    assert row["verdict"] == "HOLD-REVIEW"
    assert row["t1_blocked"] is True
    assert "major_structural_event" in row["pre_order_blockers"]


# --- CFR D5: stale 2.97% would FAIL; latest declared $1.03 -> STEP1-RECHECK ---
def test_golden_cfr_d5_recheck_not_fail():
    hist = _q(date(2023, 5, 30), 13, [1.00] * 12 + [1.03])
    row = _verdict(
        ticker="CFR",
        quote={"price": 134.70},
        profile={"sector": "Financial Services"},
        dividend_history=hist,
        floor_pct=3.0,
    )
    assert row["verdict"] == "STEP1-RECHECK"  # NOT FAIL (the D5 bug)
    assert row["t1_blocked"] is True


# --- recent cut -> FAIL (distinct from freeze) ---
def test_golden_recent_cut_fail():
    hist = _q(date(2023, 1, 1), 12, [0.50] * 8 + [0.40] * 4)
    row = _verdict(
        ticker="CUTX",
        quote={"price": 30.0},
        profile={"sector": "Industrials"},
        dividend_history=hist,
        floor_pct=3.0,
    )
    assert row["dividend_basis"]["cut_flag"] is True
    assert row["verdict"] == "FAIL"

"""WS-2 sector-aware payout-safety tests (offline, deterministic).

# DATA-DATE: 2026-05-17
# REVIEW-TRIGGER: MKC Unilever close, FITB post-Comerica normalization
"""

from payout_safety import assess_payout_safety


# --- D2: MKC GAAP EPS inflated by non-cash gain; adjusted is the anchor ---
def test_mkc_gaap_distorted_flags_one_off_uses_adjusted():
    # GAAP EPS TTM inflated to ~12.9 by a non-cash remeasurement gain;
    # adjusted ~3.09 (FY guide). Div 1.92.
    a = assess_payout_safety(
        sector="consumer staples",
        annual_dividend=1.92,
        gaap_eps=12.9,
        adjusted_eps=3.09,
        adjusted_eps_source="MANUAL",
    )
    assert a.one_off_flag is True
    assert a.gaap_adj_divergence is not None and a.gaap_adj_divergence > 0.25
    # Adjusted-EPS payout ~62%, FCF unknown -> CAUTION band, not a fake "PASS"
    assert round(a.adjusted_eps_payout, 2) == round(1.92 / 3.09, 2)
    assert a.safety_verdict in ("CAUTION", "PASS")


def test_adjusted_eps_unavailable_caps_hold_review():
    a = assess_payout_safety(
        sector="consumer",
        annual_dividend=2.0,
        gaap_eps=5.0,
        adjusted_eps=None,
        adjusted_eps_source="UNAVAILABLE",
    )
    assert "adjusted_eps_unavailable" in a.blockers
    assert a.safety_verdict == "HOLD-REVIEW"  # fail-safe, never silent PASS


# --- FITB/Comerica: completed merger -> GAAP distorted -> HOLD-REVIEW ---
def test_completed_merger_without_adjusted_holds_review():
    a = assess_payout_safety(
        sector="banks",
        annual_dividend=1.60,
        gaap_eps=0.60,  # Q1-2026 GAAP $0.15-style distortion (annualized small)
        adjusted_eps=None,
        completed_merger_within_4q=True,
    )
    assert a.safety_verdict == "HOLD-REVIEW"
    assert a.one_off_flag is True
    assert "adjusted_eps_unavailable" in a.blockers


# --- banks: FCF ignored; NPL/NCO deterioration -> CAUTION (OZK golden) ---
def test_bank_npl_nco_deterioration_caution():
    a = assess_payout_safety(
        sector="financial services",
        annual_dividend=1.88,
        gaap_eps=6.17,
        adjusted_eps=6.17,
        adjusted_eps_source="FMP",
        fcf_per_share=-50.0,  # must be ignored for a bank
        bank_metrics={
            "cet1": 0.11,
            "cre_concentration": "high",
            "npl_trend": "deteriorating",
            "nco_trend": "deteriorating",
        },
    )
    assert a.sector_kind == "bank"
    assert a.safety_verdict == "CAUTION"
    assert "bank_npl_nco_deteriorating" in a.blockers


# --- utilities: negative FCF must NOT auto-FAIL ---
def test_utility_negative_fcf_not_auto_fail():
    a = assess_payout_safety(
        sector="utilities",
        annual_dividend=1.68,
        gaap_eps=2.74,
        adjusted_eps=2.74,
        adjusted_eps_source="FMP",
        fcf_per_share=-3.0,  # structurally negative -> ignored for utilities
        utility_metrics={"ffo_to_debt": 0.16, "rate_case_status": "constructive"},
    )
    assert a.sector_kind == "utility"
    assert a.safety_verdict == "PASS"


def test_utility_missing_ffo_debt_blocks():
    a = assess_payout_safety(
        sector="utilities",
        annual_dividend=1.68,
        gaap_eps=2.74,
        adjusted_eps=2.74,
        adjusted_eps_source="FMP",
        utility_metrics={"rate_case_status": "constructive"},
    )
    assert "utility_ffo_debt_unavailable" in a.blockers
    assert a.safety_verdict in ("CAUTION", "HOLD-REVIEW")


def test_utility_adverse_rate_case_holds_review():
    a = assess_payout_safety(
        sector="utilities",
        annual_dividend=1.68,
        gaap_eps=2.74,
        adjusted_eps=2.74,
        adjusted_eps_source="FMP",
        utility_metrics={"ffo_to_debt": 0.15, "rate_case_status": "adverse"},
    )
    assert a.safety_verdict == "HOLD-REVIEW"
    assert "utility_rate_case_adverse" in a.blockers


# --- consumer: clean name passes; over-100% FCF payout fails ---
def test_consumer_clean_pass():
    a = assess_payout_safety(
        sector="consumer staples",
        annual_dividend=1.0,
        gaap_eps=3.0,
        adjusted_eps=3.0,
        adjusted_eps_source="FMP",
        fcf_per_share=2.5,
    )
    assert a.safety_verdict == "PASS"
    assert not a.blockers


def test_consumer_fcf_payout_over_100_fails():
    a = assess_payout_safety(
        sector="consumer",
        annual_dividend=3.0,
        gaap_eps=4.0,
        adjusted_eps=4.0,
        adjusted_eps_source="FMP",
        fcf_per_share=2.0,  # 3.0/2.0 = 150% FCF payout
    )
    assert a.safety_verdict == "FAIL"

"""7th-review fixes: trailing-median special, MD/CSV surfacing, safety=None."""

import csv
from datetime import date, timedelta

from build_entry_signals import build_entry_row, render_markdown, write_csv
from dividend_basis import analyze_dividends
from event_scanner import CLEAN_CONFIRMED, ScanResult
from verdict import synthesize_verdict


def _q(start, n, amt, step=91):
    return [
        {
            "date": (start + timedelta(days=step * i)).isoformat(),
            "dividend": (amt[i] if isinstance(amt, list) else amt),
            "label": "cash",
        }
        for i in range(n)
    ]


# --- Blocker1: long-term regular growth must NOT be flagged special ---
def test_long_term_regular_growth_not_marked_special():
    hist = _q(
        date(2018, 1, 1),
        16,
        [
            0.10,
            0.10,
            0.11,
            0.12,
            0.13,
            0.14,
            0.15,
            0.16,
            0.18,
            0.20,
            0.22,
            0.24,
            0.26,
            0.28,
            0.30,
            0.33,
        ],
    )
    b = analyze_dividends(hist, price=25.0, floor_pct=3.0)
    assert b.special_dividend_flag is False
    assert b.latest_declared_dividend == 0.33  # not collapsed to a stale amount


def test_true_special_still_detected_against_local_median():
    # Steady ~0.30 regular with a single 1.50 spike -> still special.
    amts = [0.30] * 11 + [1.50] + [0.30]
    b = analyze_dividends(_q(date(2022, 1, 1), 13, amts), price=40.0, floor_pct=3.0)
    assert b.special_dividend_flag is True
    assert b.variable_policy_flag is False


# --- Blocker2: Markdown + CSV surface verdict / t1_blocked / blockers ---
def _row_triggered_skipped():
    return build_entry_row(
        ticker="TRG",
        alpha_pp=0.5,
        quote={"price": 10.0},
        profile={"sector": "Industrials"},
        key_metrics=[{"dividendYield": 0.03}],
        dividend_history=_q(date(2023, 1, 1), 13, [0.10 + 0.01 * i for i in range(13)]),
        floor_pct=3.0,  # no event_scan -> SKIPPED -> HOLD-REVIEW + T1 block
    )


def test_markdown_surfaces_verdict_and_blockers():
    md = render_markdown([_row_triggered_skipped()], as_of="2026-05-17", alpha_pp=0.5)
    assert "Verdict" in md and "T1 Blocked" in md and "Pre-order Blockers" in md
    assert "HOLD-REVIEW" in md
    assert "event_scan_failed_or_skipped" in md


def test_csv_surfaces_verdict_and_blockers(tmp_path):
    p = tmp_path / "out.csv"
    write_csv([_row_triggered_skipped()], p)
    rows = list(csv.DictReader(p.open()))
    r = rows[0]
    assert r["verdict"] == "HOLD-REVIEW"
    assert r["t1_blocked"] in ("True", "true", "1")
    assert "event_scan_failed_or_skipped" in r["pre_order_blockers"]
    assert "event_scan_result" in r and "step1_verdict" in r


# --- Medium: safety_verdict None must not be CLEAN-PASS by default ---
def test_safety_none_defaults_hold_review():
    v = synthesize_verdict(
        step1_verdict="STEP1-PASS",
        safety_verdict=None,
        event_verdict_cap=None,
        pre_order_blockers=[],
    )
    assert v.verdict == "HOLD-REVIEW"
    assert "payout_safety_not_evaluated" in v.reasons


def test_safety_none_allowed_when_require_safety_false():
    v = synthesize_verdict(
        step1_verdict="STEP1-PASS",
        safety_verdict=None,
        event_verdict_cap=None,
        pre_order_blockers=[],
        require_safety=False,
    )
    assert v.verdict == "CLEAN-PASS"


def test_library_path_no_financials_not_clean_pass():
    # build_entry_row with financials omitted: Step-2 not evaluated.
    row = build_entry_row(
        ticker="LIB",
        alpha_pp=0.5,
        quote={"price": 20.0},
        profile={"sector": "Industrials"},
        key_metrics=[{"dividendYield": 0.05}],
        dividend_history=_q(date(2022, 1, 1), 13, 0.20),
        floor_pct=3.0,
        event_scan=ScanResult("LIB", CLEAN_CONFIRMED),
    )
    assert row["verdict"] != "CLEAN-PASS"

"""Exercise the real generated HTTP client in this skill's coverage row."""

import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[4]
sys.path.insert(0, str(ROOT / "scripts/tests"))
from _dividend_fmp_scenarios import run_scenarios


def test_generated_transport_and_domain_adapters():
    run_scenarios(ROOT / "skills/kanchi-dividend-sop/scripts", "build_entry_signals")

"""6th-review safety-gap fixes (offline, deterministic)."""

from datetime import date, timedelta

from build_entry_signals import build_entry_row
from event_scanner import (
    CLEAN_CONFIRMED,
    FAILED_DEGRADED,
    ScanResult,
    apply_event_cap,
)
from verdict import synthesize_verdict


def _q(start, n, amt, step=91):
    return [
        {
            "date": (start + timedelta(days=step * i)).isoformat(),
            "dividend": (amt[i] if isinstance(amt, list) else amt),
            "label": "cash",
        }
        for i in range(n)
    ]


# --- High1: Step-1 not run cannot reach a PASS tier ---
def test_high1_step1_none_is_recheck():
    v = synthesize_verdict(
        step1_verdict=None,
        safety_verdict="PASS",
        event_verdict_cap=None,
        pre_order_blockers=[],
    )
    assert v.verdict == "STEP1-RECHECK" and v.t1_blocked is True


def test_high1_unknown_step1_is_recheck():
    v = synthesize_verdict(step1_verdict="WHATEVER", safety_verdict="PASS", event_verdict_cap=None)
    assert v.verdict == "STEP1-RECHECK"


def test_high1_cli_without_yield_floor_not_pass_tier():
    row = build_entry_row(
        ticker="X",
        alpha_pp=0.5,
        quote={"price": 50.0},
        profile={"sector": "Industrials"},
        key_metrics=[{"dividendYield": 0.04}],
        dividend_history=_q(date(2023, 1, 1), 13, 0.50),
        floor_pct=None,  # Step-1 gate skipped
        event_scan=ScanResult("X", CLEAN_CONFIRMED),
    )
    assert row["verdict"] == "STEP1-RECHECK"


# --- High2: missing event scan caps a TRIGGERED name ---
def test_high2_missing_scan_on_triggered_caps_hold_review():
    # Deeply in the buy zone -> TRIGGERED; no event_scan passed at all.
    row = build_entry_row(
        ticker="TRG",
        alpha_pp=0.5,
        quote={"price": 10.0},
        profile={"sector": "Industrials"},
        key_metrics=[{"dividendYield": 0.03}],
        dividend_history=_q(date(2023, 1, 1), 13, [0.10 + 0.01 * i for i in range(13)]),
        floor_pct=3.0,
        # event_scan omitted -> treated as SKIPPED
    )
    assert row["event_scan"]["result"] == "SKIPPED"
    assert row["verdict"] == "HOLD-REVIEW"
    assert row["t1_blocked"] is True
    assert "event_scan_failed_or_skipped" in row["pre_order_blockers"]


# --- High3: completed M&A from the event scan feeds WS-2 ---
def test_high3_event_completed_mna_feeds_payout_safety():
    row = build_entry_row(
        ticker="FITB",
        alpha_pp=0.5,
        quote={"price": 40.0},
        profile={"sector": "Financial Services"},
        key_metrics=[{"dividendYield": 0.035}],
        dividend_history=_q(date(2023, 1, 1), 13, 0.35),
        floor_pct=3.0,
        financials={
            "sector": "Financial Services",
            "gaap_eps": 0.60,
            "adjusted_eps": None,
            "adjusted_eps_source": "UNAVAILABLE",
        },
        event_scan=ScanResult("FITB", CLEAN_CONFIRMED, completed_mna_within_4q=True),
    )
    # GAAP distorted by the completed merger + no adjusted EPS -> HOLD-REVIEW
    assert row["payout_safety"]["safety_verdict"] == "HOLD-REVIEW"
    assert row["payout_safety"]["one_off_flag"] is True


# --- Med1: invalid result string is pessimistic, not clean ---
def test_med1_typo_result_is_pessimistic():
    # "FAILED_DEGRADED" (underscore) is NOT the valid "FAILED-DEGRADED".
    cap = apply_event_cap(ScanResult("Z", "FAILED_DEGRADED"), step5_triggered=True)
    assert cap["verdict_cap"] == "HOLD-REVIEW"
    assert "event_scan_failed_or_skipped" in cap["blockers"]


def test_med1_unknown_result_not_clean():
    cap = apply_event_cap(ScanResult("Z", "totally-bogus"), step5_triggered=False)
    assert cap["t1_blocked"] is True  # not silently clean


def test_clean_confirmed_still_clean():
    cap = apply_event_cap(ScanResult("Z", FAILED_DEGRADED), step5_triggered=False)
    assert cap["t1_blocked"] is True
    ok = apply_event_cap(ScanResult("Z", CLEAN_CONFIRMED), step5_triggered=True)
    assert ok["verdict_cap"] is None and ok["t1_blocked"] is False


# --- 7th-review: provenance.unresolved_blockers must match the full set ---
def test_provenance_unresolved_blockers_includes_event_blockers():
    row = build_entry_row(
        ticker="TRG2",
        alpha_pp=0.5,
        quote={"price": 10.0},
        profile={"sector": "Industrials"},
        key_metrics=[{"dividendYield": 0.03}],
        dividend_history=_q(date(2023, 1, 1), 13, [0.10 + 0.01 * i for i in range(13)]),
        floor_pct=3.0,
        # no event_scan -> SKIPPED -> event_scan_failed_or_skipped blocker
    )
    assert "event_scan_failed_or_skipped" in row["pre_order_blockers"]
    # Audit provenance must report the COMPLETE set, not the verdict subset.
    assert row["provenance"]["unresolved_blockers"] == row["pre_order_blockers"]
    assert "event_scan_failed_or_skipped" in row["provenance"]["unresolved_blockers"]

"""WS-5 verdict-synthesis tests (offline, deterministic)."""

from verdict import build_run_context, evidence_ref, synthesize_verdict, worst


def test_clean_pass_when_all_green():
    v = synthesize_verdict(
        step1_verdict="STEP1-PASS",
        safety_verdict="PASS",
        event_verdict_cap=None,
        pre_order_blockers=[],
    )
    assert v.verdict == "CLEAN-PASS" and v.t1_blocked is False


def test_pass_caution_when_blockers_present():
    v = synthesize_verdict(
        step1_verdict="STEP1-PASS",
        safety_verdict="PASS",
        event_verdict_cap=None,
        pre_order_blockers=["bank_npl_nco_deteriorating"],
    )
    assert v.verdict == "PASS-CAUTION"


def test_step1_fail_is_fail():
    v = synthesize_verdict(step1_verdict="FAIL", safety_verdict="PASS", event_verdict_cap=None)
    assert v.verdict == "FAIL" and v.t1_blocked is True


def test_step1_recheck_dominates_d5():
    v = synthesize_verdict(
        step1_verdict="STEP1-RECHECK", safety_verdict="PASS", event_verdict_cap=None
    )
    assert v.verdict == "STEP1-RECHECK" and v.t1_blocked is True


def test_event_major_caps_hold_review():
    v = synthesize_verdict(
        step1_verdict="STEP1-PASS",
        safety_verdict="PASS",
        event_verdict_cap="HOLD-REVIEW",
        event_t1_blocked=True,
    )
    assert v.verdict == "HOLD-REVIEW" and v.t1_blocked is True


def test_freeze_with_clean_safety_is_conditional_pass():
    # CMCSA-style: frozen dividend but income/FCF safety strong.
    v = synthesize_verdict(
        step1_verdict="HOLD-REVIEW",
        safety_verdict="PASS",
        event_verdict_cap=None,
        pre_order_blockers=[],
    )
    assert v.verdict == "CONDITIONAL-PASS"


def test_freeze_with_blockers_is_hold_review():
    v = synthesize_verdict(
        step1_verdict="HOLD-REVIEW",
        safety_verdict="CAUTION",
        event_verdict_cap=None,
        pre_order_blockers=["adjusted_eps_unavailable"],
    )
    assert v.verdict == "HOLD-REVIEW"


def test_worst_picks_lowest_tier():
    assert worst(["CLEAN-PASS", "HOLD-REVIEW", "PASS-CAUTION"]) == "HOLD-REVIEW"
    assert worst(["CLEAN-PASS", "PASS-CAUTION"]) == "PASS-CAUTION"


def test_run_context_carries_profile_guard():
    ctx = build_run_context(
        profile="balanced",
        yield_floor_pct=3.0,
        safety_bias="medium",
        universe_source="finviz",
        excluded_asset_types=["reit", "bdc"],
    )
    assert ctx["yield_floor_pct"] == 3.0 and "bdc" in ctx["excluded_asset_types"]


def test_evidence_ref_shape():
    e = evidence_ref(
        "CMCSA dividend maintained at $1.32",
        source_type="issuer_ir",
        source_url="https://example",
        confidence="high",
    )
    assert e["claim"].startswith("CMCSA") and e["confidence"] == "high"

"""Single source of truth (SSOT) for Kanchi SOP thresholds.

Per improvement plan v2 MJ-8: thresholds live here, not duplicated across
SKILL.md / default-thresholds.md / inline script args. The markdown references
are human-readable mirrors of these values, not independent definitions.

Schema version is bumped whenever the build_entry_signals.py JSON output
contract changes (MJ-10), so downstream consumers (kanchi-dividend-review-monitor)
can detect and tolerate schema evolution.
"""

from __future__ import annotations

# JSON output contract version (WS-1 / MJ-10).
SCHEMA_VERSION = 2

# --- Step 1: yield floor + entry alpha ---
DEFAULT_ALPHA_PP = 0.5

# Profile-specific yield floors (default-thresholds.md "Objective Tuning").
YIELD_FLOOR_BY_PROFILE = {
    "income-now": 4.0,
    "balanced": 3.0,
    "growth-first": 1.5,
}

# --- Data Freshness Gate (v2.1, makes D5 explicit) ---
# If the latest-declared-dividend yield is within this band of the floor,
# the Step-1 decision is too sensitive to stale dividend data: force a
# STEP1-RECHECK (never a hard FAIL) until the latest declared dividend is
# confirmed from an authoritative source.
FLOOR_FRESHNESS_BAND_PP = 0.20

# --- WS-1: dividend-action detection ---
# A pay above this multiple of the trailing-median regular pay is treated
# as a special/supplemental dividend (amount-outlier path, label-independent).
SPECIAL_OUTLIER_MULTIPLE = 1.5
# An extreme isolated spike (>= this x trailing-local median) is a special
# on magnitude alone (e.g. ORI's ~7x annual special), even at the series
# end where no later pay exists to confirm a reversion.
SPECIAL_EXTREME_MULTIPLE = 3.0
# Between 1.5x and 3x it is a special ONLY if a later pay reverts to near
# the trailing-local median (a one-off), distinguishing it from ordinary
# steep dividend growth which does not revert.
SPECIAL_REVERT_MULTIPLE = 1.2

# Coefficient of variation of the 12-month-normalized *residual* (post-special)
# dividend stream above which the issuer is treated as a variable-dividend
# payer (e.g. CALM pays ~1/3 of net income).
VARIABLE_POLICY_COV = 0.35

# Equality epsilon for freeze/cut comparison of annualized dividend totals
# (absolute USD per share). Below this, two annual totals are "equal".
DIVIDEND_EQUALITY_EPS = 0.005

# Grace period (days) added to the inferred increase cadence before a
# same-amount window is treated as a freeze rather than "just hasn't raised
# yet this cycle" (v2.1 R-2, reduces annual-raiser false positives).
FREEZE_GRACE_DAYS = 45

# Minimum number of regular dividend pays required before statistical
# detection (CoV / freeze) is meaningful. Below this -> ASSUMPTION-REQUIRED.
MIN_REGULAR_PAYS = 4

# --- WS-2: payout triad + one-off divergence ---
ADJ_EPS_PAYOUT_CAUTION = 0.70  # 0.70-0.85 = caution band
ADJ_EPS_PAYOUT_MAX = 0.85
FCF_PAYOUT_MAX = 0.80
FCF_PAYOUT_HIGH_RISK = 1.00

# Sector-module thresholds (5th-review F2 — deepen bank/utility/insurer).
BANK_DEPOSIT_BETA_HIGH = 0.50  # deposit cost passes through fast -> margin risk
UTILITY_FFO_DEBT_MIN = 0.13  # below this, regulated leverage is stretched
INSURER_OP_EPS_PAYOUT_MAX = 0.85  # operating-EPS payout hard ceiling
# GAAP vs Adjusted EPS divergence above which a Step-4 one-off flag fires.
# Marked calibratable (MN-4): initial value, revisit with a payout-gap
# distribution study across Dividend Aristocrats.
GAAP_ADJ_DIVERGENCE = 0.25

# --- WS-3: structural-event materiality (v2.1 R-4) ---
MNA_TX_VALUE_PCT_MCAP = 0.10
MNA_SHARE_ISSUANCE_PCT = 0.10
MNA_LEVERAGE_DELTA_EBITDA = 0.5
# A completed merger within this many trailing quarters presumes GAAP EPS
# is distorted -> force the adjusted-EPS path (WS-2 linkage, FITB/Comerica).
COMPLETED_MNA_LOOKBACK_QUARTERS = 4

# --- WS-6: portfolio constraints (default-thresholds.md mirror) ---
MAX_SINGLE_POSITION_PCT = 8.0
MAX_SECTOR_PCT = 25.0
# Sector cluster-risk warning: >= this many same-sector PASS names share one
# macro beta (NEW-4, e.g. many small banks).
SECTOR_CLUSTER_WARN_COUNT = 4

# --- Verdict tier (WS-5, v2.1) ---
VERDICTS = (
    "CLEAN-PASS",
    "PASS-CAUTION",
    "CONDITIONAL-PASS",
    "HOLD-REVIEW",
    "STEP1-RECHECK",
    "FAIL",
)

"""WS-5: actionable verdict tier + provenance synthesis.

Pure, offline. Synthesizes the final tier from the WS-1 Step-1 decision,
the WS-2 payout-safety verdict, the WS-3 event cap and the aggregated
pre-order blockers. Replaces the binary signal with the reviewer-validated
actionable tier so the SOP yields a usable shortlist (Customer Issue-1),
not "always 0 PASS".

Tier (best -> worst): CLEAN-PASS, PASS-CAUTION, CONDITIONAL-PASS,
HOLD-REVIEW, STEP1-RECHECK, FAIL.
"""

from __future__ import annotations

from dataclasses import dataclass, field

from thresholds import VERDICTS  # tuple, authoritative enum

_RANK = {v: i for i, v in enumerate(VERDICTS)}  # 0 = best (CLEAN-PASS)


@dataclass
class FinalVerdict:
    verdict: str
    t1_blocked: bool
    reasons: list[str] = field(default_factory=list)


def synthesize_verdict(
    *,
    step1_verdict: str | None,
    safety_verdict: str | None,
    event_verdict_cap: str | None,
    event_t1_blocked: bool = False,
    pre_order_blockers: list[str] | None = None,
    require_safety: bool = True,
) -> FinalVerdict:
    blockers = pre_order_blockers or []
    reasons: list[str] = []

    # Hard FAILs first (irrecoverable).
    if step1_verdict == "FAIL":
        return FinalVerdict("FAIL", True, ["step1_fail"])
    if safety_verdict == "FAIL":
        return FinalVerdict("FAIL", True, ["payout_safety_fail"])

    # Step-1 is the core yield-floor gate. If it was not run (no
    # --yield-floor) or returned an unknown value, a PASS tier cannot be
    # asserted (6th-review High1): fail-safe to STEP1-RECHECK.
    _STEP1_OK = {"STEP1-PASS", "HOLD-REVIEW"}
    if step1_verdict not in _STEP1_OK and step1_verdict not in (
        "STEP1-RECHECK",
        "ASSUMPTION-REQUIRED",
    ):
        return FinalVerdict("STEP1-RECHECK", True, ["step1_not_run_or_unknown"])

    # Step-1 data-freshness recheck dominates (cannot assert PASS yet).
    if step1_verdict in ("STEP1-RECHECK", "ASSUMPTION-REQUIRED"):
        return FinalVerdict("STEP1-RECHECK", True, ["step1_recheck"])

    # Event / safety HOLD-REVIEW caps.
    if event_verdict_cap == "HOLD-REVIEW":
        reasons.append("event_cap_hold_review")
        return FinalVerdict("HOLD-REVIEW", True, reasons)
    if safety_verdict == "HOLD-REVIEW":
        reasons.append("payout_safety_hold_review")
        return FinalVerdict("HOLD-REVIEW", True, reasons)

    # Step-2 must have been evaluated to assert any PASS tier. A library
    # caller that omits financials (safety_verdict is None) cannot reach
    # CLEAN/CONDITIONAL-PASS — fail-safe to HOLD-REVIEW (7th-review Medium).
    if safety_verdict is None and require_safety:
        return FinalVerdict("HOLD-REVIEW", True, ["payout_safety_not_evaluated"])

    # Dividend freeze: income cash-cow only if safety is clean & unblocked.
    if step1_verdict == "HOLD-REVIEW":  # WS-1 emits this for freeze
        if safety_verdict == "PASS" and not blockers:
            return FinalVerdict(
                "CONDITIONAL-PASS",
                event_t1_blocked,
                ["dividend_freeze_income_cash_cow"],
            )
        return FinalVerdict("HOLD-REVIEW", True, ["dividend_freeze_unconfirmed_safety"])

    # Step-1 passed on regular yield. Blend safety + blockers.
    if safety_verdict == "CAUTION" or blockers:
        if blockers:
            reasons.append("open_pre_order_blockers")
        if safety_verdict == "CAUTION":
            reasons.append("payout_safety_caution")
        return FinalVerdict("PASS-CAUTION", event_t1_blocked, reasons)

    if safety_verdict == "PASS" and not blockers:
        return FinalVerdict("CLEAN-PASS", event_t1_blocked, ["step1_pass_safety_pass"])
    if safety_verdict is None and not blockers:
        # require_safety=False explicitly accepted (e.g. Step-5-only use).
        return FinalVerdict("CLEAN-PASS", event_t1_blocked, ["step1_pass_safety_skipped"])

    return FinalVerdict("PASS-CAUTION", event_t1_blocked, ["residual_caution"])


def worst(verdicts: list[str]) -> str:
    """Return the worst (highest-rank) verdict in a list."""
    return max(verdicts, key=lambda v: _RANK.get(v, 0)) if verdicts else "STEP1-RECHECK"


def build_run_context(
    *,
    profile: str | None,
    yield_floor_pct: float | None,
    safety_bias: str | None,
    universe_source: str | None,
    excluded_asset_types: list[str] | None,
) -> dict:
    """Top-level run_context (4th-review #13) so a 3%-run Step-1 result is
    never silently reused inside a 4%-run."""
    return {
        "profile": profile,
        "yield_floor_pct": yield_floor_pct,
        "safety_bias": safety_bias,
        "universe_source": universe_source,
        "excluded_asset_types": excluded_asset_types or [],
    }


def evidence_ref(
    claim: str,
    *,
    source_type: str,
    source_url: str | None = None,
    checked_at: str | None = None,
    raw_value=None,
    normalized_value=None,
    confidence: str = "medium",
) -> dict:
    """One evidence record for the provenance block (4th-review #8)."""
    return {
        "claim": claim,
        "source_type": source_type,
        "source_url": source_url,
        "checked_at": checked_at,
        "raw_value": raw_value,
        "normalized_value": normalized_value,
        "confidence": confidence,
    }

---
name: kanchi-dividend-sop
description: Convert Kanchi-style dividend investing into a repeatable US-stock operating procedure. Use when users ask for かんち式配当投資, dividend screening, dividend growth quality checks, PERxPBR adaptation for US sectors, pullback limit-order planning, or one-page stock memo creation. Covers screening, deep dive, entry planning, and post-purchase monitoring cadence.
---

# Kanchi Dividend Sop

## Overview

Implement Kanchi's 5-step method as a deterministic workflow for US dividend investing.
Prioritize safety and repeatability over aggressive yield chasing.

## When to Use

Use this skill when the user needs:
- Kanchi-style dividend stock selection adapted for US equities.
- A repeatable screening and pullback-entry process instead of ad-hoc picks.
- One-page underwriting memos with explicit invalidation conditions.
- A handoff package for monitoring and tax/account-location workflows.

## Prerequisites

### API Key Setup

The entry signal script requires FMP API access:

```bash
export FMP_API_KEY=your_api_key_here
```

### Input Sources

Prepare one of the following inputs before running the workflow:
1. Output from `skills/value-dividend-screener/scripts/screen_dividend_stocks.py`.
2. Output from `skills/dividend-growth-pullback-screener/scripts/screen_dividend_growth_rsi.py`.
3. User-provided ticker list (broker export or manual list).

#### Expected JSON Input Format

When using `--input`, provide JSON in one of these formats:

```json
{
  "profile": "balanced",
  "candidates": [
    {"ticker": "JNJ", "bucket": "core"},
    {"ticker": "O", "bucket": "satellite"}
  ]
}
```

Or simplified:

```json
{
  "tickers": ["JNJ", "PG", "KO"]
}
```

The optional `value-dividend-screener` and
`dividend-growth-pullback-screener` handoffs use `stocks[].symbol`.
Both `build_sop_plan.py --input` and `build_entry_signals.py --input`
accept that shape directly, as well as the native `candidates[].ticker` and
`tickers[]` shapes above.

For deterministic artifact generation, provide tickers to:

```bash
python3 skills/kanchi-dividend-sop/scripts/build_sop_plan.py \
  --tickers "JNJ,PG,KO" \
  --output-dir reports/
```

For Step 5 entry timing artifacts. **`--yield-floor` is mandatory** — it is
the Step-1 yield gate; without it every row fail-safes to `STEP1-RECHECK`
(a row can never reach a PASS tier without Step 1). Pass `--profile` /
`--safety-bias` for run_context, and `--events-json` for the Step 4b scan
(absent ⇒ every row is treated as `SKIPPED` and a TRIGGERED name is capped
to `HOLD-REVIEW` — never silently clean):

```bash
python3 skills/kanchi-dividend-sop/scripts/build_entry_signals.py \
  --tickers "JNJ,PG,KO" \
  --alpha-pp 0.5 \
  --yield-floor 3.0 \
  --profile balanced --safety-bias medium \
  --events-json reports/kanchi_events_2026-05-17.json \
  --output-dir reports/
```

## Workflow

### 1) Define mandate before screening

Collect and lock the parameters first:
- Objective: current cash income vs dividend growth.
- Max positions and position-size cap.
- Allowed instruments: stock only, or include REIT/BDC/ETF.
- Preferred account type context: taxable vs IRA-like accounts.

Load `references/default-thresholds.md` and apply baseline
settings unless the user overrides.

### 2) Build the investable universe

Start with a quality-biased universe:
- Core bucket: long dividend growth names (for example, Dividend Aristocrats style quality set).
- Satellite bucket: higher-yield sectors (utilities, telecom, REITs) in a separate risk bucket.

Use explicit source priority for ticker collection:
1. `skills/value-dividend-screener/scripts/screen_dividend_stocks.py` output (FMP/FINVIZ).
2. `skills/dividend-growth-pullback-screener/scripts/screen_dividend_growth_rsi.py` output.
3. User-provided broker export or manual ticker list when APIs are unavailable.

Return a ticker list grouped by bucket before moving forward.

### 3) Apply Kanchi Step 1 (yield filter with trap flag)

Primary rule:
- Step-1 yield = the **regular forward yield** = `latest_declared_regular
  dividend × cadence-implied frequency / price` (WS-1 `dividend_basis.py`).
  Never use `profile.lastDividend` / TTM — it lags the latest declared raise
  (defect D5) and silently bundles specials (D4).
- Apply the profile floor (income-now 4.0% / balanced 3.0% / growth-first
  1.5%) to the **regular** yield only.

Trap & freshness controls (machine-emitted by `dividend_basis.py`):
- `special_dividend_flag` → exclude specials; report regular vs ttm yield.
- `variable_policy_flag` → `FAIL` (CALM-style; not an income base).
- `cut_flag` → `FAIL`; `suspension_flag` → `FAIL`.
- `freeze_flag` → `HOLD-REVIEW` (income cash-cow exception decided in
  Step 8 synthesis only if safety is clean & unblocked).
- **Data Freshness Gate**: if the regular yield is within ±0.20pp of the
  floor (`floor_borderline`) and the latest declared dividend is not
  confirmed from an authoritative source, emit `STEP1-RECHECK` — **never a
  hard FAIL** (this is the CFR D5 fix).

### 4) Apply Kanchi Step 2 (growth and safety) — sector-dispatched

Safety is **sector-specific** — a uniform GAAP/FCF triad mis-judges banks
(FCF meaningless) and regulated utilities (FCF structurally negative).
Use `references/sector-step2-modules.md`; the deterministic dispatch is
`scripts/payout_safety.py`.

- Always compute the **payout triad**: GAAP-EPS payout, Adjusted-EPS
  payout, FCF payout. The safety verdict uses **Adjusted-EPS + FCF**
  (consumer), or the sector module (bank / utility / insurer).
- `adjusted_eps_source = UNAVAILABLE` ⇒ cap `HOLD-REVIEW` (fail-safe;
  never a silent PASS).
- GAAP↔Adjusted EPS divergence > 25% ⇒ Step-4 one-off flag.
- A merger **completed within 4 quarters** presumes GAAP EPS is distorted
  ⇒ force the adjusted path or `HOLD-REVIEW` (FITB/Comerica golden case).
- Regulated utilities: **negative FCF is not an auto-FAIL** — judge on
  FFO/debt + allowed ROE + rate-case + equity-issuance risk.

When trend is mixed but not broken, classify as `HOLD-REVIEW` instead of
hard reject.

### 5) Apply Kanchi Step 3 (valuation) with US sector mapping

Use `references/valuation-and-one-off-checks.md` and apply
sector-specific valuation logic:
- Financials: `PER x PBR` can remain primary.
- REITs: use `P/FFO` or `P/AFFO` instead of plain `P/E`.
- Asset-light sectors: combine forward `P/E`, `P/FCF`, and historical range.

Always report which valuation method was used for each ticker.

### 6) Apply Kanchi Step 4 (one-off event filter)

Reject or downgrade names where recent profits rely on one-time effects:
- Asset sale gains, litigation settlement, tax effect spikes.
- Margin spike unsupported by sales trend.
- Repeated "one-time/non-recurring" adjustments.

Record one-line evidence for each `FAIL` to keep auditability.

### 6b) Apply Kanchi Step 4b (forward structural-event scan)

Step 4 is backward-looking; Step 4b catches *pending/recent* structural
events (the MKC-Unilever miss, D3). For each surviving candidate, run a
WebSearch + issuer-IR/SEC check using the **source hierarchy**: issuer IR
→ SEC filing (8-K/10-Q/10-K/proxy/S-4) → exchange/company deck →
reputable wire → finance portals (secondary only). Record findings into a
curated events JSON and pass it via `build_entry_signals.py --events-json`.

- Only a **major structural event** caps the verdict to `HOLD-REVIEW`
  (tx > 10% mcap, share issuance > 10–20%, leverage +0.5x EBITDA,
  control/listing/HQ change, merger-of-equals / RMT / spin-off / large
  asset sale, dividend/rating/leverage-policy change, sector-specific
  materiality, or rolling-24m cumulative M&A > 15% mcap). Minor bolt-ons
  are a CAUTION note only.
- **Pessimistic cap**: `FAILED-DEGRADED` / `SKIPPED` / `NO_EVENT_FOUND`
  on a Step-5 TRIGGERED name ⇒ `HOLD-REVIEW` + **T1 BLOCKED**. WebSearch
  unavailable (web app / offline) is treated the same — never a silent
  skip. `CLEAN_CONFIRMED` (primary source checked) is stronger than
  `NO_EVENT_FOUND` (search only).

### 7) Apply Kanchi Step 5 (buy on weakness with rules)

Set entry triggers mechanically:
- Yield trigger: current yield above 5y average yield + alpha (default `+0.5pp`).
- Valuation trigger: target multiple reached (`P/E`, `P/FFO`, or `P/FCF`).

Execution pattern:
- Split orders: `40% -> 30% -> 30%`.
- **Pre-order blockers**: if a candidate has any unresolved
  `pre_order_blockers[]` (from WS-1/2/3 — variable/cut/suspension,
  adjusted-EPS-unavailable, GAAP/Adj divergence, bank credit, utility
  FFO/debt, event-scan failed/skipped, stale dividend, …) OR
  `t1_blocked` is true, the first tranche is **blocked or downsized to a
  ≤20% tracking tranche** — not 40%.
- **Sector cluster risk**: when ≥ `SECTOR_CLUSTER_WARN_COUNT` same-sector
  names pass (e.g. many small banks share one macro beta), emit a
  portfolio-level `CLUSTER-RISK` warning.
- Require one-sentence sanity check before each *unblocked* add: "thesis
  intact vs structural break".

### 8) Produce standardized outputs

Always produce:
1. Screening table with the **actionable verdict tier**: `CLEAN-PASS`,
   `PASS-CAUTION`, `CONDITIONAL-PASS`, `HOLD-REVIEW`, `STEP1-RECHECK`,
   `FAIL` (synthesized by `verdict.py` from Step 1 + Step 2 + Step 4b +
   blockers). Include evidence per row.
2. One-page stock memo (use `references/stock-note-template.md`) with the
   per-ticker **provenance block** (price/dividend/payout/event sources,
   `unresolved_blockers`, `evidence_refs[]`).
3. Limit-order plan with split sizing, blocker gate, and invalidation.
4. Top-level **run_context** (profile, yield_floor_pct, safety_bias,
   universe_source, excluded_asset_types) so a 3%-run result is never
   silently reused inside a 4%-run.

## Output

Return and/or generate:
1. SOP screening summary in markdown.
2. Underwriting memo set based on
`references/stock-note-template.md`.
3. Optional plan artifact file generated by
`skills/kanchi-dividend-sop/scripts/build_sop_plan.py` in `reports/`.
4. Optional Step 5 entry-signal artifacts generated by
`skills/kanchi-dividend-sop/scripts/build_entry_signals.py` in `reports/`.

## Cadence

Use this minimum rhythm:
- Weekly (15 min): check dividend and business-news changes only.
- Monthly (30 min): rerun screening and refresh order levels.
- Quarterly (60 min): deep safety review using latest filings/earnings.

## Multi-Skill Handoff

Run this skill first, then hand off outputs:
1. To `kanchi-dividend-review-monitor` for daily/weekly/quarterly anomaly detection.
2. To `kanchi-dividend-us-tax-accounting` for account-location and tax classification planning.

## Guardrails

- Do not issue blind buy calls without Step 4, Step 4b and safety checks.
- Do not treat high yield as value before validating coverage quality.
- Use the **regular** forward yield for Step 1, never a special/TTM-inclusive
  figure; near-floor + unconfirmed ⇒ `STEP1-RECHECK`, not FAIL.
- A failed/skipped event scan on a TRIGGERED name ⇒ `HOLD-REVIEW` + T1
  blocked. Never silently skip Step 4b.
- Keep assumptions explicit; `adjusted_eps`/data missing ⇒ fail-safe
  `HOLD-REVIEW`, never silent PASS.

## Resources

- `scripts/thresholds.py`: **single source of truth** for all SOP
  thresholds + `SCHEMA_VERSION` (downstream schema-evolution guard).
- `scripts/dividend_basis.py`: WS-1 regular/special/variable/freeze/cut +
  Data Freshness Gate engine (pure, offline).
- `scripts/payout_safety.py`: WS-2 sector-aware GAAP/Adjusted/FCF payout
  triad + completed-merger linkage.
- `scripts/event_scanner.py`: WS-3 isolated forward/recent corporate-action
  scanner + materiality gate + pessimistic cap.
- `scripts/verdict.py`: WS-5 actionable-tier synthesis + run_context +
  evidence_ref helpers.
- `scripts/build_entry_signals.py`: orchestrator (Step 5 targets + WS-1/2/3/5
  integration). Flags: `--yield-floor`, `--events-json`, `--profile`,
  `--safety-bias`, `--universe-source`.
- `scripts/build_sop_plan.py`: deterministic SOP plan scaffold generator.
- `scripts/tests/test_golden_p0.py`: **P0 merge gate** — end-to-end frozen
  verdicts for CALM/ORI/CMCSA/MKC/CFR/cut (run via `scripts/run_all_tests.sh`).
- `references/default-thresholds.md`: human-readable threshold mirror.
- `references/sector-step2-modules.md`: Step 2 safety indicators by sector.
- `references/valuation-and-one-off-checks.md`: Step 3 valuation + Step 4 one-off.
- `references/stock-note-template.md`: one-page memo + provenance block.


## Shared FMP client

FMP data access uses the bundled generated `scripts/fmp_client.py`. Existing CLI
key/environment precedence, financial calculations and successful output shapes
are preserved. The migrated paths use stable endpoints only; a denied or failed
stable request does not fall back to retired v3 URLs. The shared client masks raw
and URL-encoded key echoes, retries malformed JSON once, and retries a 429 once
with a 60-second wait before stopping on a repeated rate limit. FINVIZ, where
used, remains a separate client.

The Python client retains `sleep_seconds`, `timeout` and `api_calls`. Sleep now
sets the minimum interval between requests; accounting includes retried HTTP
responses. Quote/profile retrieval remains one symbol per request.