Ibd Distribution Day Monitor

skill_name: ibd-distribution-day-monitor
rule_version: ibd_dd_v1.0

indexes:
  - symbol: QQQ
    benchmark_name: Nasdaq Proxy
  - symbol: SPY
    benchmark_name: S&P500 Proxy

lookback_days: 80

data:
  provider: fmp           # v1: FMP only. yfinance / alpaca / polygon are out of scope.
  api_key: null           # Set via CLI --api-key or FMP_API_KEY env var.
  allow_cached_data: false

distribution_day_rule:
  min_decline_pct: -0.002
  expiration_sessions: 25
  invalidation_gain_pct: 0.05
  invalidation_price_source: high                          # "high" | "close"
  invalidation_session_scope: after_distribution_day_only  # exclude DD's own intraday high

risk_thresholds:
  caution:
    d25_count: 3
  high:
    d25_count: 5
    d15_count: 3
    d5_count: 2
  severe:
    d25_count: 6
    d15_count: 4
    severe_ma_d25: 5

moving_average_filters:
  enabled: true
  ema_periods:
    - 21
  sma_periods:
    - 50

strategy_context:
  instrument: TQQQ
  current_exposure_pct: 100
  base_trailing_stop_pct: 10
  allow_switch_to_qqq: true
  allow_hedge_recommendation: false

"""FMP ``/api/v3`` → ``/stable`` URL compatibility shim.

FMP retired the legacy ``/api/v3/`` surface on 2025-08-31; API keys issued
after that date receive ``403 "Legacy Endpoint"`` on every ``/api/v3/`` request.

This helper rewrites a legacy v3-style URL (and its params) to the ``/stable``
equivalent. It is applied ONLY at construction points that build hardcoded v3
URLs and are *not* part of an explicit stable→v3 fallback list. Methods that
already iterate a ``_FMP_ENDPOINTS`` stable→v3 table must NOT route through this
shim, or the v3 fallback entry would be rewritten back to stable and the
fallback contract would break.

Note on endpoint naming: ``/stable`` endpoint names are inconsistent. Most
legacy underscore names resolve, so unmapped endpoints fall through to a 1:1
underscore-preserving swap. But a few endpoints (``sp500_constituent`` and
``earning_calendar``) return **404 on the underscore form for all tiers** —
their live ``/stable`` name is hyphenated (verified 2026-06). Those are pinned
to the hyphenated form in ``_PATH_RENAME_NO_SYMBOL`` below. Do not "modernize"
the underscore-preserving fallthrough wholesale, and do not revert the pinned
endpoints back to underscore.
"""

from __future__ import annotations

from datetime import date, timedelta

_STABLE = "https://financialmodelingprep.com/stable"

# v3 path segment (symbol carried in the path) → /stable path (symbol via ?symbol=)
_PATH_WITH_SYMBOL = {
    "quote": "/quote",
    "profile": "/profile",
    "income-statement": "/income-statement",
    "balance-sheet-statement": "/balance-sheet-statement",
    "cash-flow-statement": "/cash-flow-statement",
    "key-metrics": "/key-metrics",
    "ratios": "/ratios",
    "enterprise-values": "/enterprise-values",
    "market-capitalization": "/market-capitalization",
    "institutional-holder": "/institutional-ownership/symbol-ownership",
    "etf-holder": "/etf-holdings",
    "rating": "/rating",
    "discounted-cash-flow": "/discounted-cash-flow",
}

# v3 path → /stable path for endpoints that carry NO path symbol and whose
# /stable name differs from the v3 name. Explicit because the underscore
# (v3-style) /stable name 404s for these; the hyphenated name is the live one
# (verified 2026-06: /stable/sp500_constituent and /stable/earning_calendar
# both 404; the hyphenated variants are the live endpoints — 200 with a Premium
# key, lower tiers may 402). These override the underscore-preserving fallthrough.
_PATH_RENAME_NO_SYMBOL = {
    "sp500_constituent": "/sp500-constituent",
    "earning_calendar": "/earnings-calendar",
}


def v3_to_stable(url: str, params: dict | None = None) -> tuple[str, dict]:
    """Rewrite a legacy FMP v3 URL to its ``/stable`` equivalent.

    No-op for URLs that do not contain ``/api/v3/``. Unmapped endpoints fall
    back to a 1:1 underscore-preserving path swap; endpoints whose underscore
    ``/stable`` form 404s are pinned to hyphen via ``_PATH_RENAME_NO_SYMBOL``.
    """
    params = {} if params is None else dict(params)

    if "/api/v3/" not in url:
        return url, params

    after = url.split("/api/v3/", 1)[1].rstrip("/")

    # historical-price-full has a dividend sub-path and a price variant
    if after.startswith("historical-price-full/stock_dividend/"):
        params["symbol"] = after[len("historical-price-full/stock_dividend/") :]
        return _STABLE + "/dividends", params
    if after.startswith("historical-price-full/"):
        params["symbol"] = after[len("historical-price-full/") :]
        # The stable EOD endpoint ignores ``timeseries``; convert to a from/to
        # range (2x calendar days covers N trading days with weekend headroom).
        timeseries = params.pop("timeseries", None)
        if timeseries:
            today = date.today()
            params.setdefault("from", (today - timedelta(days=int(timeseries) * 2)).isoformat())
            params.setdefault("to", today.isoformat())
        return _STABLE + "/historical-price-eod/full", params

    # historical/earning_calendar/{symbol} → earnings?symbol=
    if after.startswith("historical/earning_calendar/"):
        params["symbol"] = after[len("historical/earning_calendar/") :]
        return _STABLE + "/earnings", params

    # symbol-in-path endpoints → ?symbol=
    for v3_path, stable_path in _PATH_WITH_SYMBOL.items():
        if after.startswith(v3_path + "/"):
            params["symbol"] = after[len(v3_path) + 1 :]
            return _STABLE + stable_path, params
        if after == v3_path:
            return _STABLE + stable_path, params

    # Explicit hyphenated renames for symbol-less endpoints whose underscore
    # /stable form 404s (must come before the underscore-preserving fallthrough).
    if after in _PATH_RENAME_NO_SYMBOL:
        return _STABLE + _PATH_RENAME_NO_SYMBOL[after], params

    # Best-effort 1:1 swap, preserving the underscore (v3-style) name. Endpoints
    # whose underscore /stable form is known to 404 are pinned to hyphen above.
    return _STABLE + "/" + after, params

"""FMP wrapper + OHLCV quality validation.

Returns most-recent-first list[dict] (no pandas). Records data quality
issues as audit_flags / skipped_sessions for the report layer.
"""

from __future__ import annotations

from typing import Any


def normalize_history(payload: Any) -> list[dict]:
    """Coerce FMP payload variants into a flat most-recent-first list[dict].

    - dict with "historical" key (v3 shape): return payload["historical"]
    - list (stable EOD flat list, after normalizer): return as-is
    - None or unrecognized shape: return []
    """
    if payload is None:
        return []
    if isinstance(payload, dict):
        return list(payload.get("historical") or [])
    if isinstance(payload, list):
        return list(payload)
    return []


def validate_history_quality(history: list[dict]) -> tuple[list[str], list[dict]]:
    """Identify data quality problems.

    Returns (audit_flags, skipped_sessions). Each skipped session has
    {date, reason}. The caller's downstream code is expected to treat
    skipped sessions as no-op (no DD detection / enrichment for them).
    """
    flags: list[str] = []
    skipped: list[dict] = []
    for row in history:
        date = row.get("date")
        close = row.get("close")
        volume = row.get("volume")

        if close is None:
            skipped.append({"date": date, "reason": "missing_close"})
        elif close <= 0:
            skipped.append({"date": date, "reason": "invalid_close"})

        if volume is None:
            skipped.append({"date": date, "reason": "missing_volume"})
        elif volume <= 0:
            skipped.append({"date": date, "reason": "invalid_volume"})

    if skipped:
        flags.append("data_quality_warnings")
    return flags, skipped


def fetch_ohlcv(
    client: Any,
    symbol: str,
    days: int,
) -> tuple[list[dict], dict]:
    """Fetch most-recent-first OHLCV for `symbol` via the provided FMP client.

    Returns (history, audit) where audit has:
        - data_source: "fmp"
        - symbol: str
        - days_requested: int
        - sessions_loaded: int
        - audit_flags: list[str]
        - skipped_sessions: list[dict]
    """
    audit: dict = {
        "data_source": "fmp",
        "symbol": symbol,
        "days_requested": days,
        "sessions_loaded": 0,
        "audit_flags": [],
        "skipped_sessions": [],
    }

    payload = client.get_historical_prices(symbol, days=days)
    history = normalize_history(payload)

    if not history:
        audit["audit_flags"].append("no_data_returned")
        return [], audit

    flags, skipped = validate_history_quality(history)
    audit["sessions_loaded"] = len(history)
    audit["audit_flags"].extend(flags)
    audit["skipped_sessions"] = skipped
    return history, audit


def build_fmp_client(api_key: str | None = None, max_api_calls: int = 200):
    """Lazy import to avoid pulling requests in unit tests that mock the client."""
    from fmp_client import FMPClient

    return FMPClient(api_key=api_key, max_api_calls=max_api_calls)

"""IBD-style Distribution Day detection, enrichment, and counting.

All input histories must be most-recent-first (history[0] = evaluation session).
DD at history[k] has age_sessions = k. expired = age > expiration_sessions.

5% invalidation contract (C2):
- Display: high_since = max(high in history[0 : k+1]), DD day high INCLUDED.
- Invalidation: scan history[0 : k] within expiration window only,
  using rule.invalidation_price_source ("high" or "close"). DD day EXCLUDED.
- Today DD (k=0) -> no post-DD sessions -> invalidated = False, but
  high_since = DD day's intraday high (not None).
"""

from __future__ import annotations

from models import DDRecord, DistributionDayRule

EPSILON = 1e-12


def detect_distribution_days(
    effective_history: list[dict],
    rule: DistributionDayRule,
) -> tuple[list[DDRecord], list[dict]]:
    """Detect raw Distribution Days (no enrichment).

    Returns (records, skipped_sessions). Records have only detection-time fields.
    """
    records: list[DDRecord] = []
    skipped: list[dict] = []

    for i in range(len(effective_history) - 1):
        today = effective_history[i]
        yesterday = effective_history[i + 1]
        if not _has_valid_detection_pair(today, yesterday):
            skipped.append({"date": today.get("date"), "reason": "missing_or_invalid_close_volume"})
            continue

        pct_change = today["close"] / yesterday["close"] - 1
        volume_up = today["volume"] > yesterday["volume"]
        if pct_change <= rule.min_decline_pct + EPSILON and volume_up:
            volume_change_pct = today["volume"] / yesterday["volume"] - 1
            records.append(
                DDRecord(
                    date=today["date"],
                    dd_index=i,
                    age_sessions=i,
                    close=today["close"],
                    pct_change=pct_change,
                    volume=today["volume"],
                    prev_volume=yesterday["volume"],
                    volume_change_pct=volume_change_pct,
                )
            )
    return records, skipped


def enrich_records(
    records: list[DDRecord],
    effective_history: list[dict],
    rule: DistributionDayRule,
) -> list[DDRecord]:
    """Fill display, invalidation, expiration, and status fields on each record."""
    for r in records:
        k = r.dd_index

        # --- Display: high_since includes DD day (history[0 : k+1]) ---
        sessions_on_or_after_dd = effective_history[0 : k + 1]
        valid_highs = [
            row["high"]
            for row in sessions_on_or_after_dd
            if row.get("high") is not None and row["high"] > 0
        ]
        r.high_since = max(valid_highs) if valid_highs else r.close

        # --- Invalidation: scan history[0 : k] within expiration window ---
        r.invalidation_price = r.close * (1 + rule.invalidation_gain_pct)
        ev = _find_invalidation_event(effective_history, k, r.close, rule)
        r.invalidation_date = ev["date"] if ev else None
        r.invalidation_trigger_price = ev["trigger_price"] if ev else None
        r.invalidation_trigger_source = ev["trigger_source"] if ev else None

        # --- Status priority: invalidated > expired > active ---
        if ev is not None:
            r.status = "invalidated"
            r.removal_reason = "invalidated_5pct_gain"
        elif r.age_sessions > rule.expiration_sessions:
            r.status = "expired"
            r.removal_reason = "expired_25_sessions"
        else:
            r.status = "active"
            r.removal_reason = None

        r.expires_in_sessions = max(rule.expiration_sessions - r.age_sessions, 0)
    return records


def count_active_in_window(
    records: list[DDRecord],
    max_age_sessions: int,
) -> int:
    """Count active records within elapsed-session window (age <= max_age_sessions).

    Note: 'within N elapsed sessions' (age 0..N inclusive). NOT '直近 N 取引日'
    (which would be age 0..N-1).
    """
    return sum(1 for r in records if r.status == "active" and r.age_sessions <= max_age_sessions)


def _has_valid_detection_pair(today: dict, yesterday: dict) -> bool:
    """For DD detection: requires close, volume only (high not needed)."""
    for row in (today, yesterday):
        for key in ("close", "volume"):
            v = row.get(key)
            if v is None or v <= 0:
                return False
    return True


def _find_invalidation_event(
    effective_history: list[dict],
    dd_index: int,
    dd_close: float,
    rule: DistributionDayRule,
) -> dict | None:
    """Find the first chronological post-DD session where price crosses 5% threshold.

    Constraints:
    - Scans only post-DD sessions (effective_history[0 : dd_index]).
    - Restricts to expiration_sessions window: event_index >= dd_index - expiration.
    - Iterates oldest-first (chronological) to return the FIRST crossing.
    - Skips rows where row[source] is missing, None, or <= 0.
    - Today DD (dd_index <= 0) -> no scan, returns None.
    """
    if dd_index <= 0:
        return None

    source = rule.invalidation_price_source
    if source not in {"high", "close"}:
        raise ValueError(f"Unsupported invalidation_price_source: {source}")

    threshold = dd_close * (1 + rule.invalidation_gain_pct)
    min_event_index = max(dd_index - rule.expiration_sessions, 0)

    # Most-recent-first: index dd_index-1 is newest post-DD, min_event_index is oldest in scan.
    # Chronological order = oldest -> newest = min_event_index -> dd_index-1.
    for event_index in range(dd_index - 1, min_event_index - 1, -1):
        row = effective_history[event_index]
        trigger_value = row.get(source)
        if trigger_value is None or trigger_value <= 0:
            continue
        if trigger_value >= threshold:
            return {
                "date": row["date"],
                "trigger_price": trigger_value,
                "trigger_source": source,
                "elapsed_sessions_since_dd": dd_index - event_index,
            }
    return None

"""Risk-level -> portfolio action mapping for TQQQ and QQQ.

TQQQ is more aggressive (3x daily leverage), so its exposure cuts faster.
"""

from __future__ import annotations

from models import PortfolioAction

_TQQQ_POLICY = {
    "NORMAL": ("HOLD_OR_FOLLOW_BASE_STRATEGY", 100, None),
    "CAUTION": ("AVOID_NEW_ADDS", 75, 7),
    "HIGH": ("REDUCE_EXPOSURE", 50, 5),
    "SEVERE": ("CLOSE_TQQQ_OR_HEDGE", 25, 3),
}

_QQQ_POLICY = {
    "NORMAL": ("HOLD_OR_FOLLOW_BASE_STRATEGY", 100, None),
    "CAUTION": ("AVOID_NEW_ADDS", 100, 8),  # don't reduce, just stop adding
    "HIGH": ("REDUCE_EXPOSURE", 75, 6),
    "SEVERE": ("REDUCE_EXPOSURE_OR_HEDGE", 50, 5),
}

_RATIONALE = {
    "TQQQ": (
        "TQQQ targets 3x daily Nasdaq returns. Distribution Day clusters "
        "amplify drawdown risk via daily compounding, so exposure is cut "
        "faster than for unleveraged QQQ."
    ),
    "QQQ": (
        "QQQ tracks Nasdaq-100 1x. Exposure cuts are smaller than TQQQ but "
        "still respond to clustered distribution."
    ),
}

_ALTERNATIVES_TQQQ = {
    "HIGH": "SWITCH_PARTIAL_TO_QQQ",
    "SEVERE": "SWITCH_TO_QQQ_OR_CASH",
}


def generate_portfolio_action(
    risk_level: str,
    instrument: str,
    current_exposure_pct: int,
    base_trailing_stop_pct: int,
) -> PortfolioAction:
    """Return the recommended portfolio action for the given risk level."""
    instrument_upper = instrument.upper()
    policy = _TQQQ_POLICY if instrument_upper == "TQQQ" else _QQQ_POLICY
    if risk_level not in policy:
        raise ValueError(f"Unknown risk_level: {risk_level}")

    action, target, cap = policy[risk_level]
    if cap is None:
        trailing_stop = base_trailing_stop_pct
    else:
        trailing_stop = min(base_trailing_stop_pct, cap)

    alternative = _ALTERNATIVES_TQQQ.get(risk_level) if instrument_upper == "TQQQ" else None

    return PortfolioAction(
        instrument=instrument_upper,
        recommended_action=action,
        current_exposure_pct=current_exposure_pct,
        target_exposure_pct=target,
        exposure_delta_pct=target - current_exposure_pct,
        trailing_stop_pct=trailing_stop,
        alternative_action=alternative,
        rationale=_RATIONALE.get(instrument_upper, ""),
    )

#!/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 IBD Distribution Day Monitor

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
- API call budget enforcement
- Batch company profile support
- Earnings calendar and historical price fetching
"""

import os
import sys
import time
from datetime import date, timedelta
from typing import Optional

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

try:
    from _fmp_compat import v3_to_stable
except ModuleNotFoundError:  # loaded by file path (e.g. repo-level contract tests)
    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
    from _fmp_compat import v3_to_stable

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


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 = date.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 ApiCallBudgetExceeded(Exception):
    """Raised when the API call budget has been exhausted."""

    pass


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

    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

    def __init__(self, api_key: Optional[str] = None, max_api_calls: int = 200):
        self.api_key = api_key or os.getenv("FMP_API_KEY")
        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.cache = {}
        self.last_call_time = 0
        self.rate_limit_reached = False
        self.retry_count = 0
        self.max_retries = 1
        self.api_calls_made = 0
        self.max_api_calls = max_api_calls
        # 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 _rate_limited_get(
        self, url: str, params: Optional[dict] = None, quiet: bool = False
    ) -> Optional[dict]:
        """Make a rate-limited GET request with budget enforcement.

        Raises:
            ApiCallBudgetExceeded: When api_calls_made >= max_api_calls
        """
        if self.api_calls_made >= self.max_api_calls:
            raise ApiCallBudgetExceeded(
                f"API call budget exhausted: {self.api_calls_made}/{self.max_api_calls} calls used"
            )

        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=30)
            self.last_call_time = time.time()
            self.api_calls_made += 1

            if response.status_code == 200:
                self.retry_count = 0
                return response.json()
            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)
                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:
                msg = f"HTTP {response.status_code} - {response.text[:200]}"
                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:
            self._last_error = f"request exception: {e}"
            print(f"ERROR: Request exception: {e}", 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 get_earnings_calendar(self, from_date: str, to_date: str) -> Optional[list[dict]]:
        """Fetch earnings calendar for a date range.

        Args:
            from_date: Start date in YYYY-MM-DD format
            to_date: End date in YYYY-MM-DD format

        Returns:
            List of earnings event dicts or None on failure.
            Each dict contains: date, symbol, eps, epsEstimated, revenue,
            revenueEstimated, time (bmo/amc)
        """
        cache_key = f"earnings_{from_date}_{to_date}"
        if cache_key in self.cache:
            return self.cache[cache_key]

        # Hardcoded v3 URL bypasses the stable→v3 fallback list; rewrite here.
        url, params = v3_to_stable(
            f"{self.BASE_URL}/earning_calendar", {"from": from_date, "to": to_date}
        )
        data = self._rate_limited_get(url, params)
        if data:
            self.cache[cache_key] = data
        return data

    def get_company_profiles(self, symbols: list[str]) -> dict[str, dict]:
        """Fetch company profiles for multiple symbols.

        The /stable profile endpoint does not support comma-batched symbols
        (a multi-symbol request returns ``[]``), so fetch one symbol at a time.

        Args:
            symbols: List of stock symbols

        Returns:
            Dict mapping symbol -> profile dict (with marketCap, sector, etc.)
        """
        results = {}
        for symbol in symbols:
            cache_key = f"profile_{symbol}"
            if cache_key in self.cache:
                cached = self.cache[cache_key]
                if isinstance(cached, dict):
                    results[symbol] = cached
                continue

            # Hardcoded v3 URL bypasses the stable→v3 fallback list; rewrite here.
            url, params = v3_to_stable(f"{self.BASE_URL}/profile/{symbol}")
            data = self._rate_limited_get(url, params)
            if data and isinstance(data, list) and data:
                profile = data[0]
                if isinstance(profile, dict):
                    # Preserve the prior lenient behavior: a profile that omits
                    # "symbol" is still returned under the requested symbol.
                    self.cache[cache_key] = profile
                    results[profile.get("symbol", symbol)] = profile
        return results

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

        Args:
            symbol: Stock symbol
            days: Number of trading days to fetch

        Returns:
            Dict with 'symbol' and 'historical' keys, where 'historical' is a
            list of price dicts (most-recent-first) with: date, open, high, low,
            close, adjClose, volume
        """
        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:
            self.cache[cache_key] = data
        return data

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

"""Effective history rebasing for as_of evaluation.

Given a most-recent-first history and an optional as_of date, return a
slice such that effective_history[0] is the evaluation session. All
downstream modules consume effective_history without needing to know
as_of_index.
"""

from __future__ import annotations


def prepare_effective_history(
    history: list[dict],
    as_of: str | None,
    required_min_sessions: int,
) -> tuple[list[dict], dict]:
    """Rebase history so index 0 is the evaluation session.

    Args:
        history: Most-recent-first list of OHLCV dicts (history[0] = latest).
        as_of: ISO date string or None. If None, evaluation session = history[0].
            If given, the date must exist in history.
        required_min_sessions: If the resulting effective_history has fewer rows,
            an "insufficient_lookback" audit flag is appended.

    Returns:
        (effective_history, audit) where audit has:
            - as_of_resolved: str
            - sessions_available: int
            - audit_flags: list[str]

    Raises:
        ValueError: If history is empty or as_of is not found.
    """
    if not history:
        raise ValueError("history is empty; cannot prepare effective_history")

    audit: dict = {"as_of_resolved": None, "sessions_available": 0, "audit_flags": []}

    if as_of is None:
        effective = history
        audit["as_of_resolved"] = history[0]["date"]
    else:
        idx = next((i for i, row in enumerate(history) if row.get("date") == as_of), None)
        if idx is None:
            raise ValueError(f"as_of {as_of} not found in loaded history")
        effective = history[idx:]
        audit["as_of_resolved"] = as_of

    audit["sessions_available"] = len(effective)
    if len(effective) < required_min_sessions:
        audit["audit_flags"].append("insufficient_lookback")
    return effective, audit

#!/usr/bin/env python3
"""IBD Distribution Day Monitor — CLI entrypoint.

Workflow:
  1. Resolve config (default.yaml under config/, optional --config override).
  2. Fetch OHLCV for each symbol via FMP.
  3. Rebase via prepare_effective_history (as_of normalization).
  4. Detect, enrich, count active DDs (d5/d15/d25).
  5. Compute MA filters (21EMA / 50SMA) -> market_below_ma flag.
  6. Classify per-index risk and combine to overall risk.
  7. Generate portfolio action for the configured instrument.
  8. Write JSON + Markdown reports to --output-dir.

API key resolution order: --api-key > config.data.api_key > $FMP_API_KEY.
"""

from __future__ import annotations

import argparse
import os
import sys
from dataclasses import asdict
from datetime import datetime, timezone
from pathlib import Path

# pyyaml may not be installed in all envs (pyproject does include it).
import yaml  # type: ignore

# Local imports — relative to scripts/ directory.
HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))

from data_loader import build_fmp_client, fetch_ohlcv  # noqa: E402
from distribution_day_tracker import (  # noqa: E402
    count_active_in_window,
    detect_distribution_days,
    enrich_records,
)
from exposure_policy import generate_portfolio_action  # noqa: E402
from history_utils import prepare_effective_history  # noqa: E402
from math_utils import calc_ema, calc_sma  # noqa: E402
from models import DDRecord, DistributionDayRule, IndexResult, RiskThresholds  # noqa: E402
from report_generator import write_json, write_markdown  # noqa: E402
from risk_classifier import classify_risk, combine_index_risks  # noqa: E402

SKILL_ID = "ibd-distribution-day-monitor"
REPORT_PREFIX = "ibd_distribution_day_monitor"
RULE_VERSION = "ibd_dd_v1.0"
DEFAULT_CONFIG_PATH = HERE.parent / "config" / "default.yaml"


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("--symbols", default=None, help="Comma-separated, e.g. QQQ,SPY")
    p.add_argument("--lookback-days", type=int, default=None)
    p.add_argument("--instrument", default=None, help="TQQQ or QQQ")
    p.add_argument("--current-exposure", type=int, default=None)
    p.add_argument("--base-trailing-stop", type=int, default=None)
    p.add_argument("--as-of", default=None, help="YYYY-MM-DD; default = latest session")
    p.add_argument("--config", default=None)
    p.add_argument("--api-key", default=None)
    p.add_argument("--output-dir", default="reports/")
    return p.parse_args(argv)


def load_config(path: str | None) -> dict:
    config_path = Path(path) if path else DEFAULT_CONFIG_PATH
    if not config_path.exists():
        raise FileNotFoundError(f"config not found: {config_path}")
    with open(config_path, encoding="utf-8") as f:
        return yaml.safe_load(f) or {}


def resolve_api_key(cli_key: str | None, config: dict) -> str:
    """API key precedence: CLI > config.data.api_key > $FMP_API_KEY."""
    if cli_key:
        return cli_key
    cfg_key = ((config.get("data") or {}).get("api_key")) or None
    if cfg_key:
        return cfg_key
    env_key = os.getenv("FMP_API_KEY")
    if env_key:
        return env_key
    raise ValueError("FMP API key required. Pass --api-key or set FMP_API_KEY env var.")


def _build_rule(config: dict) -> DistributionDayRule:
    rule_cfg = config.get("distribution_day_rule") or {}
    return DistributionDayRule(
        min_decline_pct=rule_cfg.get("min_decline_pct", -0.002),
        expiration_sessions=rule_cfg.get("expiration_sessions", 25),
        invalidation_gain_pct=rule_cfg.get("invalidation_gain_pct", 0.05),
        invalidation_price_source=rule_cfg.get("invalidation_price_source", "high"),
    )


def _build_thresholds(config: dict) -> RiskThresholds:
    t_cfg = config.get("risk_thresholds") or {}
    high = t_cfg.get("high") or {}
    severe = t_cfg.get("severe") or {}
    return RiskThresholds(
        caution_d25=(t_cfg.get("caution") or {}).get("d25_count", 3),
        high_d25=high.get("d25_count", 5),
        high_d15=high.get("d15_count", 3),
        high_d5=high.get("d5_count", 2),
        severe_d25=severe.get("d25_count", 6),
        severe_d15=severe.get("d15_count", 4),
        severe_ma_d25=severe.get("severe_ma_d25", 5),
    )


def _required_min_sessions(config: dict, rule: DistributionDayRule) -> int:
    lookback = config.get("lookback_days", 80)
    ma_cfg = config.get("moving_average_filters") or {}
    sma_periods = ma_cfg.get("sma_periods") or []
    ema_periods = ma_cfg.get("ema_periods") or []
    return max(
        int(lookback),
        max(sma_periods, default=0),
        max(ema_periods, default=0),
        rule.expiration_sessions + 2,
    )


def _compute_ma_filters(effective_history: list[dict], ma_cfg: dict) -> tuple[dict, list[str]]:
    """Compute close_above_21ema, close_above_50sma, market_below_21ema_or_50ma."""
    audit_flags: list[str] = []
    if not ma_cfg.get("enabled", True) or not effective_history:
        return {
            "close_above_21ema": None,
            "close_above_50sma": None,
            "market_below_21ema_or_50ma": None,
        }, audit_flags

    closes = [row["close"] for row in effective_history]
    today_close = closes[0]

    ema_period = (ma_cfg.get("ema_periods") or [21])[0]
    sma_period = (ma_cfg.get("sma_periods") or [50])[0]

    def _ma_or_none(fn, period):
        if len(closes) < period:
            return None
        try:
            return fn(closes, period)
        except ValueError:
            return None

    ema_val = _ma_or_none(calc_ema, ema_period)
    sma_val = _ma_or_none(calc_sma, sma_period)

    close_above_ema = (today_close > ema_val) if ema_val is not None else None
    close_above_sma = (today_close > sma_val) if sma_val is not None else None

    if close_above_ema is None or close_above_sma is None:
        audit_flags.append("insufficient_data_for_moving_average")
        market_below = None
    else:
        # market_below = True only when close is below BOTH MAs
        market_below = (not close_above_ema) and (not close_above_sma)

    return {
        "close_above_21ema": close_above_ema,
        "close_above_50sma": close_above_sma,
        "market_below_21ema_or_50ma": market_below,
    }, audit_flags


def _record_to_dict(r: DDRecord) -> dict:
    return asdict(r)


def _today_dict(effective_history: list[dict], records: list[DDRecord]) -> dict:
    """Return summary fields for the evaluation session."""
    if not effective_history:
        return {}
    today = effective_history[0]
    yesterday = effective_history[1] if len(effective_history) > 1 else None
    pct_change = None
    volume_change_pct = None
    if yesterday and today.get("close") and yesterday.get("close"):
        pct_change = today["close"] / yesterday["close"] - 1
    if yesterday and today.get("volume") and yesterday.get("volume"):
        volume_change_pct = today["volume"] / yesterday["volume"] - 1
    return {
        "date": today.get("date"),
        "close": today.get("close"),
        "previous_close": yesterday.get("close") if yesterday else None,
        "pct_change": pct_change,
        "volume": today.get("volume"),
        "previous_volume": yesterday.get("volume") if yesterday else None,
        "volume_change_pct": volume_change_pct,
    }


def _build_explanation(symbol: str, d5: int, d15: int, d25: int, risk: str, today_dd: bool) -> str:
    parts = []
    parts.append(
        f"{symbol}は本日{'Distribution Day該当' if today_dd else 'Distribution Day非該当'}。"
    )
    parts.append(f"5/15/25セッション経過以内の有効Distribution Dayはそれぞれ {d5}/{d15}/{d25} 件。")
    parts.append(f"リスク判定: {risk}。")
    return " ".join(parts)


def _build_cluster_state(d5: int, d15: int, d25: int) -> dict:
    return {
        "has_d5_cluster": d5 >= 2,
        "has_d15_cluster": d15 >= 3,
        "has_d25_cluster": d25 >= 5,
        "cluster_description": (f"5/15/25セッション経過以内: {d5}/{d15}/{d25}"),
    }


def analyze_index(
    symbol: str,
    benchmark_name: str,
    history: list[dict],
    config: dict,
    rule: DistributionDayRule,
    thresholds: RiskThresholds,
    as_of: str | None,
) -> tuple[IndexResult, list[str]]:
    """Run the full per-index pipeline and return the IndexResult."""
    required_min = _required_min_sessions(config, rule)
    effective_history, hist_audit = prepare_effective_history(history, as_of, required_min)
    audit_flags = list(hist_audit.get("audit_flags", []))

    raw_records, skipped = detect_distribution_days(effective_history, rule)
    records = enrich_records(raw_records, effective_history, rule)

    active = [r for r in records if r.status == "active"]
    removed = [r for r in records if r.status != "active"]

    d5 = count_active_in_window(records, 5)
    d15 = count_active_in_window(records, 15)
    d25 = count_active_in_window(records, 25)

    ma_cfg = config.get("moving_average_filters") or {}
    trend_filters, ma_flags = _compute_ma_filters(effective_history, ma_cfg)
    audit_flags.extend(ma_flags)

    market_below = trend_filters.get("market_below_21ema_or_50ma")
    risk = classify_risk(d5, d15, d25, market_below, thresholds)

    today_is_dd = any(r.dd_index == 0 and r.status == "active" for r in records)
    explanation = _build_explanation(symbol, d5, d15, d25, risk, today_is_dd)

    result = IndexResult(
        symbol=symbol,
        benchmark_name=benchmark_name,
        is_distribution_day_today=today_is_dd,
        today=_today_dict(effective_history, records),
        d5_count=d5,
        d15_count=d15,
        d25_count=d25,
        active_distribution_days=[_record_to_dict(r) for r in active],
        removed_distribution_days=[_record_to_dict(r) for r in removed],
        risk_level=risk,
        cluster_state=_build_cluster_state(d5, d15, d25),
        trend_filters=trend_filters,
        explanation=explanation,
        skipped_sessions=skipped,
    )
    return result, audit_flags


def build_payload(
    config: dict,
    rule: DistributionDayRule,
    thresholds: RiskThresholds,
    index_results: list[IndexResult],
    overall_risk: str,
    portfolio_action: dict,
    aggregate_audit: dict,
) -> dict:
    """Assemble the final JSON payload."""
    primary = next((r.symbol for r in index_results if r.symbol == "QQQ"), None)
    if primary is None and index_results:
        primary = index_results[0].symbol

    return {
        "market_distribution_state": {
            "as_of": aggregate_audit.get("as_of_resolved"),
            "generated_at": aggregate_audit.get("generated_at"),
            "overall_risk_level": overall_risk,
            "primary_signal_symbol": primary,
            "index_results": [_index_result_to_dict(r) for r in index_results],
        },
        "portfolio_action": portfolio_action,
        "rule_evaluation": {
            "rule_version": RULE_VERSION,
            "distribution_day_rule": {
                "min_decline_pct": rule.min_decline_pct,
                "expiration_sessions": rule.expiration_sessions,
                "invalidation_gain_pct": rule.invalidation_gain_pct,
                "invalidation_price_source": rule.invalidation_price_source,
            },
            "thresholds_used": {
                "caution_d25": thresholds.caution_d25,
                "high_d25": thresholds.high_d25,
                "high_d15": thresholds.high_d15,
                "high_d5": thresholds.high_d5,
                "severe_d25": thresholds.severe_d25,
                "severe_d15": thresholds.severe_d15,
                "severe_ma_d25": thresholds.severe_ma_d25,
            },
        },
        "audit": aggregate_audit,
    }


def _index_result_to_dict(r: IndexResult) -> dict:
    return asdict(r)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    config = load_config(args.config)

    # CLI overrides
    if args.symbols:
        config["symbols_override"] = [s.strip() for s in args.symbols.split(",") if s.strip()]
    if args.lookback_days is not None:
        config["lookback_days"] = args.lookback_days
    if args.instrument:
        config.setdefault("strategy_context", {})["instrument"] = args.instrument
    if args.current_exposure is not None:
        config.setdefault("strategy_context", {})["current_exposure_pct"] = args.current_exposure
    if args.base_trailing_stop is not None:
        config.setdefault("strategy_context", {})["base_trailing_stop_pct"] = (
            args.base_trailing_stop
        )

    # Symbols / benchmark mapping
    indexes_cfg: list[dict] = config.get("indexes") or [
        {"symbol": "QQQ", "benchmark_name": "Nasdaq Proxy"},
        {"symbol": "SPY", "benchmark_name": "S&P500 Proxy"},
    ]
    if config.get("symbols_override"):
        wanted = {s.upper() for s in config["symbols_override"]}
        indexes_cfg = [i for i in indexes_cfg if i["symbol"].upper() in wanted] or [
            {"symbol": s, "benchmark_name": f"{s} proxy"} for s in config["symbols_override"]
        ]

    # FMP client
    api_key = resolve_api_key(args.api_key, config)
    client = build_fmp_client(api_key=api_key)

    rule = _build_rule(config)
    thresholds = _build_thresholds(config)
    lookback = int(config.get("lookback_days", 80))

    aggregate_audit_flags: list[str] = []
    symbols_loaded: list[str] = []
    skipped_sessions_all: list[dict] = []
    index_results: list[IndexResult] = []
    as_of_resolved: str | None = args.as_of

    for entry in indexes_cfg:
        symbol = entry["symbol"]
        benchmark = entry.get("benchmark_name") or f"{symbol} Proxy"
        history, fetch_audit = fetch_ohlcv(client, symbol, days=lookback + 5)
        aggregate_audit_flags.extend(fetch_audit.get("audit_flags", []))
        skipped_sessions_all.extend(fetch_audit.get("skipped_sessions", []))

        if not history:
            continue
        symbols_loaded.append(symbol)

        result, idx_flags = analyze_index(
            symbol=symbol,
            benchmark_name=benchmark,
            history=history,
            config=config,
            rule=rule,
            thresholds=thresholds,
            as_of=args.as_of,
        )
        aggregate_audit_flags.extend(idx_flags)
        skipped_sessions_all.extend(result.skipped_sessions)
        # Preserve as_of from the first successful index.
        if as_of_resolved is None and result.today.get("date"):
            as_of_resolved = result.today["date"]
        index_results.append(result)

    if not index_results:
        print("ERROR: no symbols loaded", file=sys.stderr)
        return 1

    overall_risk = combine_index_risks(index_results)

    strategy_ctx = config.get("strategy_context") or {}
    instrument = strategy_ctx.get("instrument", "TQQQ")
    current_exposure = int(strategy_ctx.get("current_exposure_pct", 100))
    base_trail = int(strategy_ctx.get("base_trailing_stop_pct", 10))
    portfolio_action_obj = generate_portfolio_action(
        risk_level=overall_risk,
        instrument=instrument,
        current_exposure_pct=current_exposure,
        base_trailing_stop_pct=base_trail,
    )

    generated_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
    aggregate_audit = {
        "data_source": "fmp",
        "rule_version": RULE_VERSION,
        "as_of_resolved": as_of_resolved,
        "lookback_days": lookback,
        "symbols_requested": [i["symbol"] for i in indexes_cfg],
        "symbols_loaded": symbols_loaded,
        "skipped_sessions": skipped_sessions_all,
        "audit_flags": sorted(set(aggregate_audit_flags)),
        "generated_at": generated_at,
        "config_snapshot": {
            "data": config.get("data"),
            "lookback_days": lookback,
            "moving_average_filters": config.get("moving_average_filters"),
            "strategy_context": strategy_ctx,
        },
    }

    payload = build_payload(
        config=config,
        rule=rule,
        thresholds=thresholds,
        index_results=index_results,
        overall_risk=overall_risk,
        portfolio_action=asdict(portfolio_action_obj),
        aggregate_audit=aggregate_audit,
    )

    output_dir = Path(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    timestamp = datetime.now().strftime("%Y-%m-%d_%H%M%S")
    json_path = output_dir / f"{REPORT_PREFIX}_{timestamp}.json"
    md_path = output_dir / f"{REPORT_PREFIX}_{timestamp}.md"
    write_json(payload, json_path)
    write_markdown(payload, md_path)
    print(f"Wrote {json_path}")
    print(f"Wrote {md_path}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""
Shared EMA/SMA calculation utilities (copied from market-top-detector).

All functions expect prices in most-recent-first order.
"""


def calc_ema(prices: list[float], period: int) -> float:
    """
    Calculate Exponential Moving Average from prices (most recent first).

    Args:
        prices: List of prices, most recent first.
        period: EMA period (must be >= 1).

    Returns:
        EMA value as float.

    Raises:
        ValueError: If prices is empty or period < 1.
    """
    if not prices:
        raise ValueError("prices must not be empty")
    if period < 1:
        raise ValueError("period must be >= 1")

    if len(prices) < period:
        return sum(prices) / len(prices)

    prices_rev = prices[::-1]
    sma = sum(prices_rev[:period]) / period
    ema = sma
    k = 2 / (period + 1)
    for p in prices_rev[period:]:
        ema = p * k + ema * (1 - k)
    return ema


def calc_sma(prices: list[float], period: int) -> float:
    """
    Calculate Simple Moving Average from prices (most recent first).

    Args:
        prices: List of prices, most recent first.
        period: SMA period (must be >= 1).

    Returns:
        SMA value as float.

    Raises:
        ValueError: If prices is empty or period < 1.
    """
    if not prices:
        raise ValueError("prices must not be empty")
    if period < 1:
        raise ValueError("period must be >= 1")

    if len(prices) < period:
        return sum(prices) / len(prices)
    return sum(prices[:period]) / period

"""Data models for IBD Distribution Day Monitor.

All OHLCV history is stored as a list[dict] in most-recent-first order
(history[0] = latest session). pandas is intentionally not used.
"""

from __future__ import annotations

from dataclasses import dataclass, field


@dataclass
class DistributionDayRule:
    """IBD-style Distribution Day detection rule."""

    min_decline_pct: float = -0.002
    expiration_sessions: int = 25
    invalidation_gain_pct: float = 0.05
    invalidation_price_source: str = "high"  # "high" | "close"


@dataclass
class RiskThresholds:
    """Configurable thresholds for NORMAL/CAUTION/HIGH/SEVERE classification."""

    caution_d25: int = 3
    high_d25: int = 5
    high_d15: int = 3
    high_d5: int = 2
    severe_d25: int = 6
    severe_d15: int = 4
    severe_ma_d25: int = 5  # SEVERE escalation when market_below_ma is True


@dataclass
class DDRecord:
    """A single Distribution Day record with enrichment fields."""

    date: str
    dd_index: int  # = age_sessions, index in effective_history
    age_sessions: int
    close: float
    pct_change: float
    volume: int
    prev_volume: int
    volume_change_pct: float
    # Filled by enrich_records:
    high_since: float | None = None
    invalidation_price: float | None = None
    invalidation_date: str | None = None
    invalidation_trigger_price: float | None = None
    invalidation_trigger_source: str | None = None
    expires_in_sessions: int | None = None
    status: str = "active"  # active | expired | invalidated
    removal_reason: str | None = None


@dataclass
class IndexResult:
    """Per-index analysis result."""

    symbol: str
    benchmark_name: str
    is_distribution_day_today: bool
    today: dict
    d5_count: int
    d15_count: int
    d25_count: int
    active_distribution_days: list[dict]
    removed_distribution_days: list[dict]
    risk_level: str
    cluster_state: dict
    trend_filters: dict
    explanation: str
    skipped_sessions: list[dict] = field(default_factory=list)


@dataclass
class PortfolioAction:
    """Recommended exposure action for the configured instrument."""

    instrument: str
    recommended_action: str
    current_exposure_pct: int
    target_exposure_pct: int
    exposure_delta_pct: int
    trailing_stop_pct: int
    alternative_action: str | None = None
    rationale: str = ""

"""Report generation for IBD Distribution Day Monitor.

- UTF-8 only (encoding="utf-8")
- JSON uses ensure_ascii=False so Japanese explanations are preserved as-is.
- Sensitive keys are redacted via lowercase comparison (H4).
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

# H4: lowercase only. Compared via k.lower() in SENSITIVE_KEYS.
SENSITIVE_KEYS = {
    "api_key",
    "apikey",
    "fmp_api_key",
    "secret",
    "client_secret",
    "token",
    "access_token",
    "refresh_token",
    "authorization",
    "password",
}
REDACTED = "***REDACTED***"


def _redact(obj: Any) -> Any:
    if isinstance(obj, dict):
        return {
            k: (REDACTED if isinstance(k, str) and k.lower() in SENSITIVE_KEYS else _redact(v))
            for k, v in obj.items()
        }
    if isinstance(obj, list):
        return [_redact(x) for x in obj]
    return obj


def write_json(payload: dict, path: str | Path) -> None:
    safe = _redact(payload)
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(safe, f, ensure_ascii=False, indent=2, default=str)


def write_markdown(payload: dict, path: str | Path) -> None:
    safe = _redact(payload)
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    text = render_markdown(safe)
    with open(path, "w", encoding="utf-8") as f:
        f.write(text)


def render_markdown(payload: dict) -> str:
    """Render the analysis payload as a human-readable markdown report."""
    state = payload.get("market_distribution_state", {})
    action = payload.get("portfolio_action", {})
    audit = payload.get("audit", {})

    lines: list[str] = []
    lines.append("# IBD Distribution Day Monitor Report")
    lines.append("")
    lines.append(f"- **As of:** {state.get('as_of', 'N/A')}")
    lines.append(f"- **Overall Risk Level:** **{state.get('overall_risk_level', 'N/A')}**")
    lines.append(f"- **Primary Signal Symbol:** {state.get('primary_signal_symbol', 'N/A')}")
    lines.append(f"- **Generated At:** {state.get('generated_at', 'N/A')}")
    lines.append("")

    lines.append("## Index Results")
    lines.append("")
    for idx in state.get("index_results", []) or []:
        lines.append(f"### {idx.get('symbol')} — risk: **{idx.get('risk_level')}**")
        lines.append("")
        lines.append(f"- Today is Distribution Day: {idx.get('is_distribution_day_today')}")
        lines.append(
            f"- d5 / d15 / d25 = {idx.get('d5_count')} / {idx.get('d15_count')} / {idx.get('d25_count')}"
        )
        cluster = idx.get("cluster_state") or {}
        if cluster:
            lines.append(f"- Cluster: {cluster.get('cluster_description', '')}")
        trend = idx.get("trend_filters") or {}
        if trend:
            lines.append(
                "- Trend filters: "
                f"close_above_21ema={trend.get('close_above_21ema')}, "
                f"close_above_50sma={trend.get('close_above_50sma')}, "
                f"market_below_21ema_or_50ma={trend.get('market_below_21ema_or_50ma')}"
            )
        explanation = idx.get("explanation") or ""
        if explanation:
            lines.append("")
            lines.append(f"> {explanation}")

        active = idx.get("active_distribution_days") or []
        if active:
            lines.append("")
            lines.append("#### Active Distribution Days")
            lines.append("")
            lines.append(
                "| date | close | pct_change | volume_change_pct | age | expires_in |"
                " high_since | invalidation_price |"
            )
            lines.append(
                "|------|-------|------------|-------------------|-----|------------|"
                "------------|---------------------|"
            )
            for r in active:
                lines.append(
                    "| {date} | {close} | {pc} | {vc} | {age} | {exp} | {hs} | {inv} |".format(
                        date=r.get("date"),
                        close=r.get("close"),
                        pc=_fmt_pct(r.get("pct_change")),
                        vc=_fmt_pct(r.get("volume_change_pct")),
                        age=r.get("age_sessions"),
                        exp=r.get("expires_in_sessions"),
                        hs=r.get("high_since"),
                        inv=r.get("invalidation_price"),
                    )
                )
        lines.append("")

    lines.append("## Portfolio Action")
    lines.append("")
    if action:
        lines.append(f"- **Instrument:** {action.get('instrument')}")
        lines.append(f"- **Recommended Action:** {action.get('recommended_action')}")
        lines.append(
            "- **Exposure:** "
            f"current {action.get('current_exposure_pct')}% → "
            f"target {action.get('target_exposure_pct')}% "
            f"(delta {action.get('exposure_delta_pct')}%)"
        )
        lines.append(f"- **Trailing Stop:** {action.get('trailing_stop_pct')}%")
        if action.get("alternative_action"):
            lines.append(f"- **Alternative:** {action.get('alternative_action')}")
        rationale = action.get("rationale") or ""
        if rationale:
            lines.append("")
            lines.append(f"> {rationale}")
    lines.append("")

    lines.append("## Audit")
    lines.append("")
    lines.append(f"- **Data Source:** {audit.get('data_source', 'N/A')}")
    lines.append(f"- **Symbols:** {', '.join(audit.get('symbols_loaded', []) or [])}")
    lines.append(f"- **Audit Flags:** {audit.get('audit_flags', [])}")
    lines.append(f"- **Rule Version:** {audit.get('rule_version', 'N/A')}")
    lines.append("")
    return "\n".join(lines) + "\n"


def _fmt_pct(value) -> str:
    if value is None:
        return "—"
    try:
        return f"{value * 100:.2f}%"
    except (TypeError, ValueError):
        return str(value)

"""Risk classification for individual indexes and TQQQ-aware combination."""

from __future__ import annotations

from models import IndexResult, RiskThresholds

_RISK_ORDER = {"NORMAL": 0, "CAUTION": 1, "HIGH": 2, "SEVERE": 3}


def classify_risk(
    d5: int,
    d15: int,
    d25: int,
    market_below_ma: bool | None,
    thresholds: RiskThresholds,
) -> str:
    """Classify a single index's risk level.

    market_below_ma may be True | False | None. None means MA could not be
    computed (insufficient data); SEVERE escalation is skipped in that case.
    """
    if d25 >= thresholds.severe_d25 or d15 >= thresholds.severe_d15:
        return "SEVERE"
    if market_below_ma is True and d25 >= thresholds.severe_ma_d25:
        return "SEVERE"
    if d25 >= thresholds.high_d25 or d15 >= thresholds.high_d15 or d5 >= thresholds.high_d5:
        return "HIGH"
    if d25 >= thresholds.caution_d25:
        return "CAUTION"
    return "NORMAL"


def combine_index_risks(results: list[IndexResult]) -> str:
    """Combine per-index risks into an overall risk level.

    Policy (TQQQ-aware, QQQ-weighted):
      - any SEVERE -> SEVERE
      - QQQ HIGH -> HIGH
      - QQQ NORMAL + SPY HIGH -> HIGH (broad-market degradation spills into TQQQ)
      - QQQ CAUTION + SPY in {CAUTION, HIGH} -> HIGH
      - otherwise: max risk across all indexes
    """
    if any(r.risk_level == "SEVERE" for r in results):
        return "SEVERE"

    qqq = next((r for r in results if r.symbol == "QQQ"), None)
    spy = next((r for r in results if r.symbol == "SPY"), None)

    if qqq and qqq.risk_level == "HIGH":
        return "HIGH"

    if qqq and spy:
        if qqq.risk_level == "NORMAL" and spy.risk_level == "HIGH":
            return "HIGH"
        if qqq.risk_level == "CAUTION" and spy.risk_level in ("CAUTION", "HIGH"):
            return "HIGH"

    return max((r.risk_level for r in results), key=lambda lv: _RISK_ORDER[lv])

"""Shared fixtures for IBD Distribution Day Monitor tests."""

import os
import sys

# Add scripts directory to path so modules can be imported.
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
# Add tests directory to path so helpers can be imported.
sys.path.insert(0, os.path.dirname(__file__))

"""Synthetic OHLCV builders for tests.

All builders return list[dict] in most-recent-first order
(history[0] = latest session).
"""

from __future__ import annotations

from datetime import date, timedelta


def make_bar(
    d: str,
    close: float,
    volume: int = 1_000_000,
    open_: float | None = None,
    high: float | None = None,
    low: float | None = None,
):
    """Build a single OHLCV bar dict."""
    return {
        "date": d,
        "open": open_ if open_ is not None else close,
        "high": high if high is not None else close + 0.5,
        "low": low if low is not None else close - 0.5,
        "close": close,
        "volume": volume,
    }


def date_seq(start: str, n: int, step_days: int = 1):
    """Yield n date strings going backwards from `start` (most-recent-first).

    Example: date_seq("2026-04-30", 3) -> ["2026-04-30", "2026-04-29", "2026-04-28"]
    """
    base = date.fromisoformat(start)
    return [(base - timedelta(days=i * step_days)).isoformat() for i in range(n)]


def make_history(closes: list[float], volumes: list[int] | None = None, start: str = "2026-04-30"):
    """Build a most-recent-first history from close/volume lists.

    closes[0] is the latest close. Default volume is 1_000_000.
    """
    if volumes is None:
        volumes = [1_000_000] * len(closes)
    if len(volumes) != len(closes):
        raise ValueError("closes and volumes must have the same length")
    dates = date_seq(start, len(closes))
    return [make_bar(d, c, v) for d, c, v in zip(dates, closes, volumes)]


def make_dd_history(
    *,
    dd_age: int,
    pre_dd_close: float = 100.0,
    dd_drop_pct: float = -0.0075,
    dd_volume_increase_pct: float = 0.20,
    sessions_after_dd_close: list[float] | None = None,
    sessions_after_dd_high: list[float] | None = None,
    start: str = "2026-04-30",
    base_volume: int = 1_000_000,
):
    """Build a history that contains a single Distribution Day at age=dd_age.

    Returned list is most-recent-first. effective_history[dd_age] is the DD bar.
    `sessions_after_dd_close` / `sessions_after_dd_high` describe (most-recent-first)
    the dd_age post-DD sessions; if None, they all stay flat at dd_close.
    """
    dd_close = pre_dd_close * (1 + dd_drop_pct)
    pre_dd_volume = base_volume
    dd_volume = int(pre_dd_volume * (1 + dd_volume_increase_pct))

    n_after = dd_age
    if sessions_after_dd_close is None:
        sessions_after_dd_close = [dd_close] * n_after
    if sessions_after_dd_high is None:
        sessions_after_dd_high = [c + 0.5 for c in sessions_after_dd_close]
    if len(sessions_after_dd_close) != n_after or len(sessions_after_dd_high) != n_after:
        raise ValueError("sessions_after_dd_* lengths must equal dd_age")

    bars: list[dict] = []
    dates = date_seq(start, n_after + 2)

    # Most-recent-first: post-DD sessions, then DD, then pre-DD baseline
    for i in range(n_after):
        bars.append(
            make_bar(
                dates[i],
                close=sessions_after_dd_close[i],
                volume=base_volume,
                high=sessions_after_dd_high[i],
            )
        )
    bars.append(
        make_bar(
            dates[n_after],
            close=dd_close,
            volume=dd_volume,
            high=pre_dd_close * 0.999,  # DD intraday high typically below pre-DD close
        )
    )
    bars.append(make_bar(dates[n_after + 1], close=pre_dd_close, volume=pre_dd_volume))
    return bars