Parabolic Short Trade Planner

JNJ
PG
KO
PEP
WMT
MCD
VZ
KMB
CL
MO
MMM
GIS
DUK
SO
ED

AAPL
MSFT
NVDA
AMZN
GOOG
META
JPM
XOM
V
UNH

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

"""Alpaca implementation of the broker short-inventory adapter.

Uses ``requests`` against the public Alpaca Asset endpoint
(``GET /v2/assets/{symbol}``) so this skill stays compatible with the
existing portfolio-manager skill (which already takes the same approach)
and avoids adding the ``alpaca-py`` SDK as a dependency.

Important Alpaca-specific facts that the contract reflects:

- New shorts are only permitted on **ETB (Easy-To-Borrow)** names.
  Hard-to-borrow names cannot be opened on Alpaca regardless of locate.
- Alpaca does NOT publish a borrow-fee schedule via API. ETB names are
  effectively 0% but anything not ETB is "manual check required".
- Alpaca does not support self-service locate. A non-ETB name simply
  cannot be opened, so ``manual_locate_required`` is always True (the
  trader must confirm at the broker even for ETB symbols).
"""

from __future__ import annotations

import os
from datetime import datetime, timezone

try:
    import requests
except ImportError as e:  # pragma: no cover - environment check
    raise RuntimeError("alpaca_inventory_adapter requires the `requests` package") from e

from broker_short_inventory_adapter import (
    BrokerNotConfiguredError,
    BrokerShortInventoryAdapter,
)

PAPER_BASE_URL = "https://paper-api.alpaca.markets"
LIVE_BASE_URL = "https://api.alpaca.markets"


class AlpacaInventoryAdapter(BrokerShortInventoryAdapter):
    """Read-only adapter that calls ``/v2/assets/{symbol}`` on Alpaca."""

    def __init__(
        self,
        api_key: str | None = None,
        secret_key: str | None = None,
        paper: bool = True,
        timeout: float = 10.0,
    ) -> None:
        self.api_key = api_key or os.getenv("ALPACA_API_KEY")
        self.secret_key = secret_key or os.getenv("ALPACA_SECRET_KEY")
        if not self.api_key or not self.secret_key:
            raise BrokerNotConfiguredError(
                "ALPACA_API_KEY and ALPACA_SECRET_KEY must be set (env vars or constructor args)."
            )
        self.paper = paper
        self.base_url = PAPER_BASE_URL if paper else LIVE_BASE_URL
        self.timeout = timeout

    def _headers(self) -> dict[str, str]:
        return {
            "APCA-API-KEY-ID": self.api_key,
            "APCA-API-SECRET-KEY": self.secret_key,
        }

    def get_inventory_status(self, symbol: str) -> dict:
        url = f"{self.base_url}/v2/assets/{symbol}"
        response = requests.get(url, headers=self._headers(), timeout=self.timeout)

        # Paper accounts have a smaller asset universe than live; missing
        # tickers must degrade to a blocking-but-non-fatal status so a
        # single unknown symbol cannot abort an entire Phase 2 batch.
        if response.status_code == 404:
            return {
                "shortable": False,
                "easy_to_borrow": False,
                "can_open_new_short": False,
                "borrow_fee_apr": None,
                "borrow_fee_manual_check_required": True,
                "manual_locate_required": True,
                "source": "alpaca_v2_assets",
                "checked_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
                "error": "asset_not_found",
            }

        response.raise_for_status()
        data = response.json()

        shortable = bool(data.get("shortable"))
        etb = bool(data.get("easy_to_borrow"))
        # Alpaca: new shorts only on ETB names. Treat HTB-but-shortable as
        # ``not openable`` so Phase 2 surfaces a blocking reason.
        can_open = shortable and etb
        return {
            "shortable": shortable,
            "easy_to_borrow": etb,
            "can_open_new_short": can_open,
            "borrow_fee_apr": 0.0 if etb else None,
            "borrow_fee_manual_check_required": not etb,
            "manual_locate_required": True,
            "source": "alpaca_v2_assets",
            "checked_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        }

"""Alpaca implementation of MarketDataAdapter.

Reuses the same auth pattern and 404-graceful contract as
``alpaca_inventory_adapter.py``. The market data host is
``data.alpaca.markets`` for **both** paper and live accounts —
Alpaca's account class only affects the trading API host, not the
market data API.

Free paper accounts get the IEX feed (~15 min delay). The adapter
defaults to ``feed=iex``; pass ``feed='sip'`` if you have a paid
subscription.
"""

from __future__ import annotations

import logging
import os
from datetime import datetime, time, timedelta

try:
    import requests
except ImportError as e:  # pragma: no cover - environment check
    raise RuntimeError("alpaca_market_data_adapter requires the `requests` package") from e

from broker_short_inventory_adapter import BrokerNotConfiguredError
from market_clock import (
    ET,
    REGULAR_CLOSE_HOUR,
    REGULAR_CLOSE_MINUTE,
    REGULAR_OPEN_HOUR,
    REGULAR_OPEN_MINUTE,
    to_utc,
)
from market_data_adapter import MarketDataAdapter

DATA_BASE_URL = "https://data.alpaca.markets"
TIMEFRAME = "5Min"
BAR_DURATION = timedelta(minutes=5)

logger = logging.getLogger("parabolic_short.alpaca_market_data")


class AlpacaMarketDataAdapter(MarketDataAdapter):
    def __init__(
        self,
        api_key: str | None = None,
        secret_key: str | None = None,
        paper: bool = True,
        feed: str = "iex",
        timeout: float = 15.0,
    ) -> None:
        self.api_key = api_key or os.getenv("ALPACA_API_KEY")
        self.secret_key = secret_key or os.getenv("ALPACA_SECRET_KEY")
        if not self.api_key or not self.secret_key:
            raise BrokerNotConfiguredError(
                "ALPACA_API_KEY and ALPACA_SECRET_KEY must be set (env vars or constructor args)."
            )
        # Stored for symmetry with the trading adapter; market data
        # endpoint is the same for both account classes.
        self.paper = paper
        self.feed = feed
        self.timeout = timeout
        self.base_url = DATA_BASE_URL

    def _headers(self) -> dict[str, str]:
        return {
            "APCA-API-KEY-ID": self.api_key,
            "APCA-API-SECRET-KEY": self.secret_key,
        }

    def get_bars_5min(
        self,
        symbol: str,
        *,
        session_date: str,
        until_et: datetime,
    ) -> list[dict]:
        if until_et.tzinfo is None:
            raise ValueError("until_et must be timezone-aware")

        # Build the regular-session window in ET, convert to RFC3339 UTC.
        date_obj = datetime.strptime(session_date, "%Y-%m-%d").date()
        open_et = datetime.combine(
            date_obj, time(REGULAR_OPEN_HOUR, REGULAR_OPEN_MINUTE), tzinfo=ET
        )
        close_et = datetime.combine(
            date_obj, time(REGULAR_CLOSE_HOUR, REGULAR_CLOSE_MINUTE), tzinfo=ET
        )
        # End at the earlier of close_et and until_et — there's no point
        # asking Alpaca for bars we'd then discard.
        end_et = min(close_et, until_et)
        if end_et <= open_et:
            return []

        params_base = {
            "timeframe": TIMEFRAME,
            "start": _rfc3339_utc(open_et),
            "end": _rfc3339_utc(end_et),
            "adjustment": "raw",
            "feed": self.feed,
        }

        url = f"{self.base_url}/v2/stocks/{symbol}/bars"
        all_wire_bars: list[dict] = []
        page_token: str | None = None

        while True:
            params = dict(params_base)
            if page_token is not None:
                params["page_token"] = page_token

            response = requests.get(
                url, headers=self._headers(), params=params, timeout=self.timeout
            )

            if response.status_code == 404:
                logger.info(
                    "alpaca.assets_404: %s not in Alpaca asset universe; returning [] bars",
                    symbol,
                )
                return []

            response.raise_for_status()
            payload = response.json()
            page_bars = payload.get("bars") or []
            all_wire_bars.extend(page_bars)
            page_token = payload.get("next_page_token")
            if not page_token:
                break

        return _convert_and_filter(all_wire_bars, session_date=session_date, until_et=until_et)


def _rfc3339_utc(ts_et: datetime) -> str:
    """Convert an ET datetime to ``YYYY-MM-DDTHH:MM:SSZ`` (UTC)."""
    utc = to_utc(ts_et)
    return utc.strftime("%Y-%m-%dT%H:%M:%SZ")


def _convert_and_filter(
    wire_bars: list[dict],
    *,
    session_date: str,
    until_et: datetime,
) -> list[dict]:
    """Normalise Alpaca's wire shape and apply the contract filters.

    Alpaca's bar timestamp ``t`` is the **bar-open** instant (start of
    the 5-minute interval). A bar with ``t = 09:35:00`` covers
    09:35–09:40 and is **not yet confirmed at 09:35** — it confirms at
    09:40 when the next bar starts. The contract Phase 3 needs is
    "only evaluate confirmed bars", so we filter on ``bar_close =
    bar_start + 5 min`` instead of ``bar_start <= until_et``. ``ts_et``
    in the output dict is kept as the bar-open time (the convention the
    rest of the FSM uses for transition timestamps), with the explicit
    documented meaning of "the start of the bar that triggered the
    transition" (i.e. the 5-min interval whose close fired the move).
    """
    open_et = datetime.combine(
        datetime.strptime(session_date, "%Y-%m-%d").date(),
        time(REGULAR_OPEN_HOUR, REGULAR_OPEN_MINUTE),
        tzinfo=ET,
    )
    close_et = datetime.combine(
        datetime.strptime(session_date, "%Y-%m-%d").date(),
        time(REGULAR_CLOSE_HOUR, REGULAR_CLOSE_MINUTE),
        tzinfo=ET,
    )

    out: list[dict] = []
    for wire in wire_bars:
        ts_utc_str = wire["t"]
        # Alpaca returns "...Z" — fromisoformat in 3.11+ handles "Z",
        # but be defensive across Python versions.
        if ts_utc_str.endswith("Z"):
            ts_utc_str = ts_utc_str[:-1] + "+00:00"
        ts_utc = datetime.fromisoformat(ts_utc_str)
        ts_et = ts_utc.astimezone(ET)
        bar_close_et = ts_et + BAR_DURATION

        # Regular-session filter: 09:30 ≤ bar_start < 16:00 ET on the
        # requested session_date.
        if ts_et.date().isoformat() != session_date:
            continue
        if not (open_et <= ts_et < close_et):
            continue
        # Confirmation filter: only include bars whose CLOSE
        # (bar_start + 5 min) is at or before until_et.
        if bar_close_et > until_et:
            continue

        out.append(
            {
                "ts_et": ts_et.isoformat(),
                "o": float(wire["o"]),
                "h": float(wire["h"]),
                "l": float(wire["l"]),
                "c": float(wire["c"]),
                "v": int(wire["v"]),
            }
        )
    return out

"""Fixture-driven MarketDataAdapter for offline testing.

A fixture file is a JSON object mapping ticker symbol → list of bars:

    {
      "AAPL": [
        {"ts_et": "2026-05-05T09:30:00-04:00",
         "o": 150.0, "h": 150.4, "l": 149.8, "c": 150.2, "v": 1200000},
        ...
      ],
      "NVDA": [...]
    }

Bars in the fixture should be pre-sorted chronological and use
**bar-open** semantics for ``ts_et`` (matching Alpaca wire). The
adapter only returns *confirmed* bars: a bar with ``ts_et = T``
covers ``[T, T+5min)`` and is confirmed at ``T+5min``, so it is
included only when ``T + 5min <= until_et``.
"""

from __future__ import annotations

import json
from datetime import datetime, timedelta
from pathlib import Path

from market_data_adapter import MarketDataAdapter

BAR_DURATION = timedelta(minutes=5)


class FixtureBarsAdapter(MarketDataAdapter):
    def __init__(self, fixture_path: str | Path) -> None:
        self._path = Path(fixture_path)
        self._cache: dict[str, list[dict]] | None = None

    def _load(self) -> dict[str, list[dict]]:
        if self._cache is None:
            with self._path.open(encoding="utf-8") as fh:
                self._cache = json.load(fh)
            if not isinstance(self._cache, dict):
                raise ValueError(
                    f"Fixture {self._path} must be a JSON object "
                    f"{{ticker: [bars]}}, got {type(self._cache).__name__}"
                )
        return self._cache

    def get_bars_5min(
        self,
        symbol: str,
        *,
        session_date: str,
        until_et: datetime,
    ) -> list[dict]:
        if until_et.tzinfo is None:
            raise ValueError("until_et must be timezone-aware")

        all_bars = self._load().get(symbol, [])
        if not all_bars:
            return []

        out: list[dict] = []
        for bar in all_bars:
            ts = datetime.fromisoformat(bar["ts_et"])
            if ts.tzinfo is None:
                # Defensive: a fixture writer might forget the offset.
                raise ValueError(f"Fixture bar for {symbol} is missing tz: {bar['ts_et']}")
            # Filter to the requested session date (ET wall-clock) and to
            # bars that have CLOSED at or before until_et (bar_open + 5
            # min — Alpaca-compatible bar-open semantics, see module
            # docstring).
            if ts.date().isoformat() != session_date:
                continue
            bar_close = ts + BAR_DURATION
            if bar_close > until_et:
                continue
            out.append(bar)
        return out

"""Abstract base for intraday bar fetchers used by Phase 3.

Two concrete implementations live alongside this file:

- ``FixtureBarsAdapter`` reads a JSON file mapping ``ticker -> [bars]``
  and is used in every unit test.
- ``AlpacaMarketDataAdapter`` calls
  ``data.alpaca.markets/v2/stocks/{symbol}/bars`` and is the live data
  source.

The Phase 3 FSM evaluators are pure functions — they never touch an
adapter directly. The CLI (`monitor_intraday_trigger.py`) instantiates
the right adapter based on ``--bars-source`` and passes the bar list
into ``intraday_state_machine.step_one_plan``.
"""

from __future__ import annotations

from abc import ABC, abstractmethod
from datetime import datetime


class MarketDataAdapter(ABC):
    """Contract every Phase 3 bar fetcher must satisfy."""

    @abstractmethod
    def get_bars_5min(
        self,
        symbol: str,
        *,
        session_date: str,
        until_et: datetime,
    ) -> list[dict]:
        """Return 5-min bars for the regular session of ``session_date``.

        Returned bars MUST be:
          - chronological (oldest first),
          - all from the regular cash session (09:30–16:00 ET) of
            ``session_date``,
          - filtered to **confirmed** bars only: a bar with ``ts_et =
            T`` covers ``[T, T+5min)`` and is confirmed at ``T+5min``,
            so the bar is included iff ``T + 5min <= until_et``.
            Implementations MUST NOT include unconfirmed (still-open)
            bars even if Alpaca returns them with the current minute's
            timestamp,
          - in this dict shape exactly:
              {"ts_et": <ISO 8601 with America/New_York tz; bar-open>,
               "o": float, "h": float, "l": float, "c": float,
               "v": int}

        Implementations MUST return ``[]`` (not raise) when:
          - the symbol is not in the universe (e.g. delisted, paper
            account doesn't have it),
          - the session hasn't opened yet,
          - it's a weekend / market holiday.

        ``session_date`` is the ET wall-clock date (``YYYY-MM-DD``),
        NOT a UTC date. Use ``market_clock.session_date_for(now_et)``
        to compute it.
        """
        raise NotImplementedError

"""Bar order normalization — the single boundary between raw FMP output
and calculator input.

FMP's ``historical-price-eod/full`` returns rows in most-recent-first order.
Calculators in this skill expect chronological (oldest-first) input. Call
:func:`normalize_bars` once at the screen entry point so each calculator can
assume a consistent contract.

Other responsibilities handled here (kept on the boundary, not inside the
HTTP client):

- De-duplicate rows that share the same ``date``. The latest occurrence wins
  (a re-fetch arriving after a partial bar is preferred over the partial).
- Optionally warn when consecutive calendar days are missing. This is not an
  error (weekends/holidays are normal) — it's just a hook for callers that
  want to surface unusual gaps.
"""

from __future__ import annotations

import warnings
from datetime import date
from typing import Literal

OutputOrder = Literal["chronological", "recent_first"]


def normalize_bars(
    bars: list[dict],
    output_order: OutputOrder = "chronological",
    *,
    warn_on_gaps: bool = False,
) -> list[dict]:
    """Return ``bars`` sorted in ``output_order`` with duplicates removed.

    Args:
        bars: Iterable of dicts that each carry a ``date`` field formatted
            as ``YYYY-MM-DD`` (the shape FMP returns).
        output_order: ``"chronological"`` (oldest first) or ``"recent_first"``
            (newest first). The argument names the *output*, not the input,
            to remove ambiguity at call sites.
        warn_on_gaps: When ``True``, emit a :class:`UserWarning` for every
            calendar-day gap larger than three days (ignores typical
            weekends).

    Returns:
        A new list — the input is not mutated.

    Raises:
        ValueError: If ``output_order`` is not one of the supported values.
    """
    if output_order not in ("chronological", "recent_first"):
        raise ValueError(
            f"output_order must be 'chronological' or 'recent_first', got {output_order!r}"
        )

    by_date: dict[str, dict] = {}
    for row in bars:
        d = row.get("date")
        if not d:
            continue
        # Last occurrence wins so a fresh re-fetch beats a stale partial bar.
        by_date[d] = row

    chronological = sorted(by_date.values(), key=lambda r: r["date"])

    if warn_on_gaps and len(chronological) > 1:
        for prev, curr in zip(chronological, chronological[1:]):
            try:
                d_prev = date.fromisoformat(prev["date"])
                d_curr = date.fromisoformat(curr["date"])
            except (TypeError, ValueError):
                continue
            gap = (d_curr - d_prev).days
            if gap > 3:
                warnings.warn(
                    f"bar_normalizer: gap of {gap} calendar days between "
                    f"{prev['date']} and {curr['date']}",
                    stacklevel=2,
                )

    if output_order == "chronological":
        return chronological
    return list(reversed(chronological))

"""Broker-agnostic short-inventory interface.

Phase 2 needs four answers from the broker before it can render a trade
plan: can we open a new short, what is the borrow fee, do we know the
fee, and is a manual locate needed. Different brokers expose this with
very different APIs (Alpaca: shortable / easy_to_borrow flags;
Interactive Brokers: locate API + borrow rate sheet), so the screener
talks to a thin adapter layer instead of a specific broker SDK.

Contract:

    {
        "shortable": bool,
        "easy_to_borrow": bool,
        "can_open_new_short": bool,           # alias for shortable AND ETB
        "borrow_fee_apr": float | None,       # 0.0 for ETB on Alpaca; None
                                              # if the broker can't quote
        "borrow_fee_manual_check_required": bool,
        "manual_locate_required": bool,       # always True for Alpaca short
        "source": str,                        # e.g. "alpaca_v2_assets"
        "checked_at": str,                    # ISO 8601 UTC
    }

Adapters MUST raise ``BrokerNotConfiguredError`` if their credentials are
missing — the CLI uses that to flip to ``--broker none`` (manual checklist)
instead of crashing.
"""

from __future__ import annotations

from abc import ABC, abstractmethod


class BrokerNotConfiguredError(RuntimeError):
    """Raised when a broker adapter is invoked without required credentials."""


class BrokerShortInventoryAdapter(ABC):
    """Interface every broker adapter must implement."""

    @abstractmethod
    def get_inventory_status(self, symbol: str) -> dict:
        """Return the contract dict described in the module docstring."""
        raise NotImplementedError

    def can_open_new_short(self, symbol: str) -> bool:
        """Convenience: True iff ``get_inventory_status`` says yes."""
        return bool(self.get_inventory_status(symbol).get("can_open_new_short"))


class ManualBrokerAdapter(BrokerShortInventoryAdapter):
    """Sentinel adapter used when ``--broker none`` is passed.

    Returns a dict that flags every gate as ``manual_check_required`` so
    the Phase 2 plan output explicitly tells the trader to verify locate
    and shortability at the broker before entering.
    """

    def get_inventory_status(self, symbol: str) -> dict:
        from datetime import datetime, timezone

        return {
            "shortable": None,
            "easy_to_borrow": None,
            "can_open_new_short": False,  # default-deny: tell Phase 2 to block
            "borrow_fee_apr": None,
            "borrow_fee_manual_check_required": True,
            "manual_locate_required": True,
            "source": "manual",
            "checked_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        }

"""Short-term acceleration metrics for parabolic detection.

Captures three ideas Qullamaggie flags as "exhaustion":

- ``return_n_days_pct``: total return over a recent N-day window. Inputs
  use chronological order so the latest close is at the end.
- ``consecutive_green_days``: how many consecutive bars closed up over
  the most recent run. A streak of 3-5+ is the canonical Parabolic Short
  setup.
- ``acceleration_ratio``: average daily return over the last 3 sessions
  divided by the average daily return over the last 10 sessions. Values
  > 1.5 mean the move is *accelerating* — the curve is bending up, not
  just trending.
"""

from __future__ import annotations


def return_pct(closes: list[float], days: int) -> float | None:
    """Total return over the trailing ``days`` bars, in percent.

    Returns ``None`` if there are not enough closes (we need at least
    ``days + 1`` to compute a return). Returns ``None`` if the reference
    close is non-positive.
    """
    if days <= 0:
        raise ValueError(f"days must be positive, got {days}")
    if len(closes) < days + 1:
        return None
    base = closes[-days - 1]
    if base <= 0:
        return None
    latest = closes[-1]
    return (latest - base) / base * 100.0


def consecutive_green_days(opens: list[float], closes: list[float]) -> int:
    """Count consecutive bars with ``close > open`` from the latest backwards."""
    if len(opens) != len(closes) or not closes:
        return 0
    streak = 0
    for o, c in zip(reversed(opens), reversed(closes)):
        if c > o:
            streak += 1
        else:
            break
    return streak


def _avg_daily_return(closes: list[float], window: int) -> float | None:
    if len(closes) < window + 1:
        return None
    base = closes[-window - 1]
    if base <= 0:
        return None
    # Geometric-ish: simple average of daily simple returns (sufficient for ranking)
    rets = []
    for i in range(-window, 0):
        prev = closes[i - 1]
        curr = closes[i]
        if prev <= 0:
            continue
        rets.append((curr - prev) / prev)
    if not rets:
        return None
    return sum(rets) / len(rets)


def acceleration_ratio(
    closes: list[float], short_window: int = 3, long_window: int = 10
) -> float | None:
    """Ratio of recent average daily return over a longer-window average.

    Returns ``None`` if either window cannot be computed or the long-window
    average is non-positive (no acceleration if the move was flat / down).
    """
    short_avg = _avg_daily_return(closes, short_window)
    long_avg = _avg_daily_return(closes, long_window)
    if short_avg is None or long_avg is None or long_avg <= 0:
        return None
    return short_avg / long_avg


def calculate_acceleration(
    opens: list[float],
    closes: list[float],
) -> dict:
    """Aggregate acceleration metrics into one dict for downstream scoring."""
    return {
        "return_3d_pct": return_pct(closes, 3),
        "return_5d_pct": return_pct(closes, 5),
        "return_10d_pct": return_pct(closes, 10),
        "return_15d_pct": return_pct(closes, 15),
        "consecutive_green_days": consecutive_green_days(opens, closes),
        "acceleration_ratio_3_over_10": acceleration_ratio(closes, 3, 10),
    }

"""ATR (Average True Range) and True Range helpers.

Adapted from skills/vcp-screener/scripts/calculators/vcp_pattern_calculator.py
(``_calculate_atr`` / ``_true_range``). Inputs are always in chronological
order (oldest first) — the bar_normalizer enforces this contract upstream.
"""

from __future__ import annotations


def true_range(high: float, low: float, prev_close: float) -> float:
    """Single-bar True Range.

    TR = max(high - low, |high - prev_close|, |low - prev_close|).
    """
    return max(high - low, abs(high - prev_close), abs(low - prev_close))


def calculate_atr(
    highs: list[float],
    lows: list[float],
    closes: list[float],
    period: int = 14,
) -> float:
    """Average True Range over the most recent ``period`` bars.

    Args:
        highs: High prices in chronological order.
        lows: Low prices in chronological order.
        closes: Close prices in chronological order.
        period: ATR window in bars (default 14).

    Returns:
        ATR value, or ``0.0`` if there are fewer than ``period + 1`` bars.
    """
    n = len(highs)
    if n < period + 1 or len(lows) != n or len(closes) != n:
        return 0.0

    true_ranges = [true_range(highs[i], lows[i], closes[i - 1]) for i in range(1, n)]
    if len(true_ranges) < period:
        return 0.0
    return sum(true_ranges[-period:]) / period

"""Average daily dollar volume (ADV) and a log-scale liquidity score.

ADV is the universe-side hard filter — anything below the floor is rejected
in ``invalidation_rules`` before scoring runs. The score returned here is
only used for the 10-point Liquidity factor inside the composite score.
"""

from __future__ import annotations

from math_helpers import log10_scale


def adv_dollars(closes: list[float], volumes: list[float], period: int = 20) -> float | None:
    """Average ``close * volume`` over the trailing ``period`` bars."""
    if period <= 0:
        raise ValueError(f"period must be positive, got {period}")
    if len(closes) < period or len(volumes) < period or len(closes) != len(volumes):
        return None
    pairs = list(zip(closes[-period:], volumes[-period:]))
    total = sum(c * v for c, v in pairs)
    return total / period


def latest_volume_ratio(volumes: list[float], period: int = 20) -> float | None:
    """Latest bar volume divided by the trailing-period average volume."""
    if len(volumes) < period + 1:
        return None
    avg = sum(volumes[-period - 1 : -1]) / period
    if avg <= 0:
        return None
    return volumes[-1] / avg


def calculate_liquidity(
    closes: list[float],
    volumes: list[float],
    period: int = 20,
    score_lo_log10: float = 7.0,
    score_hi_log10: float = 8.5,
) -> dict:
    """Aggregate liquidity metrics + log-scale score.

    Score endpoints default to ``$10M ADV → 0`` and ``$316M ADV → 10``.
    """
    adv = adv_dollars(closes, volumes, period=period)
    score = log10_scale(adv, score_lo_log10, score_hi_log10) if adv is not None else 0.0
    return {
        "adv_20d_usd": adv,
        "volume_ratio_20d": latest_volume_ratio(volumes, period=period),
        "liquidity_score_0_to_10": score,
    }

"""Distance of the latest close from key SMAs.

Returns three measurements that together describe how stretched a parabolic
candidate has become from its trend:

- ``ext_*_pct``: percentage distance from the SMA, signed (positive when
  price is above the MA, negative below).
- ``ext_20dma_atr``: same distance for the 20-DMA expressed in ATR(14)
  units. Volatility-adjusted comparison across symbols.

All inputs must be in chronological order (oldest first).
"""

from __future__ import annotations

from atr_calculator import calculate_atr
from math_helpers import sma


def calculate_ma_extension(
    closes: list[float],
    highs: list[float | None] = None,
    lows: list[float | None] = None,
    atr_period: int = 14,
) -> dict:
    """Compute MA-extension metrics for the most recent close.

    Args:
        closes: Closing prices (chronological).
        highs: Optional high prices for ATR-unit calculation. If not
            provided, ``ext_20dma_atr`` is returned as ``None``.
        lows: Optional low prices, see ``highs``.
        atr_period: ATR window for the volatility-adjusted distance.

    Returns:
        Dict with keys ``ext_10dma_pct``, ``ext_20dma_pct``,
        ``ext_50dma_pct``, ``ext_20dma_atr``, ``close``, ``dma_10``,
        ``dma_20``, ``dma_50``, ``atr_14``. Any value that cannot be
        computed (insufficient history) is ``None``.
    """
    if not closes:
        return {
            "close": None,
            "dma_10": None,
            "dma_20": None,
            "dma_50": None,
            "ext_10dma_pct": None,
            "ext_20dma_pct": None,
            "ext_50dma_pct": None,
            "atr_14": None,
            "ext_20dma_atr": None,
        }

    close = closes[-1]
    dma_10 = sma(closes, 10)
    dma_20 = sma(closes, 20)
    dma_50 = sma(closes, 50)

    def _pct(ma: float | None) -> float | None:
        if ma is None or ma == 0:
            return None
        return (close - ma) / ma * 100.0

    atr_14: float | None = None
    ext_20dma_atr: float | None = None
    if highs is not None and lows is not None and dma_20 is not None:
        atr_value = calculate_atr(highs, lows, closes, period=atr_period)
        if atr_value > 0:
            atr_14 = atr_value
            ext_20dma_atr = (close - dma_20) / atr_value

    return {
        "close": close,
        "dma_10": dma_10,
        "dma_20": dma_20,
        "dma_50": dma_50,
        "ext_10dma_pct": _pct(dma_10),
        "ext_20dma_pct": _pct(dma_20),
        "ext_50dma_pct": _pct(dma_50),
        "atr_14": atr_14,
        "ext_20dma_atr": ext_20dma_atr,
    }

"""Aggregate the 5 component scores (each 0-100) used by the composite scorer.

Each component takes the raw metrics computed in Day 1 and maps them to a
single 0-100 sub-score. The composite scorer in ``scorer.py`` then applies
weights (30/25/20/15/10) and converts the result to a letter grade.

These mappings are intentionally simple piecewise-linear functions so they
are predictable and easy to test. Tuning is left to ``screen_parabolic.py``
CLI knobs (e.g. ``--min-roc-5d``).
"""

from __future__ import annotations

from acceleration_calculator import calculate_acceleration
from liquidity_metrics_calculator import calculate_liquidity
from ma_extension_calculator import calculate_ma_extension
from range_expansion_calculator import calculate_range_expansion


def _clamp(value: float, lo: float = 0.0, hi: float = 100.0) -> float:
    return max(lo, min(hi, value))


def _linear(x: float | None, x_lo: float, x_hi: float) -> float:
    """Map x linearly from [x_lo, x_hi] -> [0, 100], clamped at the edges."""
    if x is None:
        return 0.0
    if x <= x_lo:
        return 0.0
    if x >= x_hi:
        return 100.0
    return (x - x_lo) / (x_hi - x_lo) * 100.0


def score_ma_extension(metrics: dict) -> float:
    """Use the larger of (20DMA % distance, 20DMA ATR distance)."""
    pct = metrics.get("ext_20dma_pct") or 0.0
    atr_units = metrics.get("ext_20dma_atr") or 0.0
    pct_score = _linear(pct, 20.0, 120.0)
    atr_score = _linear(atr_units, 3.0, 10.0)
    # Combine: 50/50 — both signals matter. ATR-units matters for vol-adjustment;
    # raw % matters for absolute climax.
    return _clamp(0.5 * pct_score + 0.5 * atr_score)


def score_acceleration(metrics: dict) -> float:
    r5 = metrics.get("return_5d_pct") or 0.0
    r3 = metrics.get("return_3d_pct") or 0.0
    streak = metrics.get("consecutive_green_days") or 0
    accel = metrics.get("acceleration_ratio_3_over_10") or 0.0

    return_score = _linear(r5, 20.0, 100.0)
    streak_score = _linear(streak, 2, 5)
    accel_score = _linear(accel, 1.0, 2.0)
    short_burst_score = _linear(r3, 10.0, 50.0)
    return _clamp(
        0.4 * return_score + 0.2 * streak_score + 0.2 * accel_score + 0.2 * short_burst_score
    )


def score_volume_climax(volume_ratio: float | None) -> float:
    return _linear(volume_ratio, 1.5, 5.0)


def score_range_expansion(expansion_ratio: float | None) -> float:
    return _linear(expansion_ratio, 1.2, 3.0)


def score_liquidity(liquidity_score_0_to_10: float) -> float:
    """Map the 0-10 ADV log-scale score onto 0-100."""
    return _clamp(liquidity_score_0_to_10 * 10.0)


def calculate_component_scores(
    closes: list[float],
    opens: list[float],
    highs: list[float],
    lows: list[float],
    volumes: list[float],
) -> dict:
    """Compute the 5 component sub-scores (each 0-100) plus raw metrics.

    Used by ``scorer.py`` and exposed in the screener output for transparency.
    """
    ma_metrics = calculate_ma_extension(closes=closes, highs=highs, lows=lows)
    accel_metrics = calculate_acceleration(opens=opens, closes=closes)
    range_metrics = calculate_range_expansion(highs=highs, lows=lows, closes=closes)
    liq_metrics = calculate_liquidity(closes=closes, volumes=volumes)

    return {
        "components": {
            "ma_extension": score_ma_extension(ma_metrics),
            "acceleration": score_acceleration(accel_metrics),
            "volume_climax": score_volume_climax(liq_metrics.get("volume_ratio_20d")),
            "range_expansion": score_range_expansion(range_metrics.get("expansion_ratio")),
            "liquidity": score_liquidity(liq_metrics.get("liquidity_score_0_to_10", 0.0)),
        },
        "raw_metrics": {
            **{k: v for k, v in ma_metrics.items()},
            **{k: v for k, v in accel_metrics.items()},
            **{k: v for k, v in range_metrics.items()},
            **{k: v for k, v in liq_metrics.items()},
        },
    }

"""Range-expansion metric: latest 5-day average True Range divided by the
prior 20-day ATR.

A value > 2.0 means the current move's daily range has roughly doubled
versus the prior month's volatility — the canonical "blow-off" signature
for parabolic exhaustion.
"""

from __future__ import annotations

from atr_calculator import true_range


def calculate_range_expansion(
    highs: list[float],
    lows: list[float],
    closes: list[float],
    short_window: int = 5,
    long_window: int = 20,
) -> dict:
    """Latest short-window ATR / prior long-window ATR.

    The two windows are non-overlapping by construction:
    - ``recent_avg_tr`` averages the last ``short_window`` true ranges
    - ``baseline_avg_tr`` averages the ``long_window`` true ranges that
      precede that recent block.

    Returns dict with ``recent_avg_tr``, ``baseline_avg_tr``, and
    ``expansion_ratio``. Any value that cannot be computed is ``None``.
    """
    n = len(highs)
    if n < short_window + long_window + 1 or len(lows) != n or len(closes) != n:
        return {
            "recent_avg_tr": None,
            "baseline_avg_tr": None,
            "expansion_ratio": None,
        }

    trs = [true_range(highs[i], lows[i], closes[i - 1]) for i in range(1, n)]
    recent = trs[-short_window:]
    baseline = trs[-(short_window + long_window) : -short_window]
    recent_avg = sum(recent) / short_window
    baseline_avg = sum(baseline) / long_window

    expansion_ratio: float | None = None
    if baseline_avg > 0:
        expansion_ratio = recent_avg / baseline_avg

    return {
        "recent_avg_tr": recent_avg,
        "baseline_avg_tr": baseline_avg,
        "expansion_ratio": expansion_ratio,
    }

#!/usr/bin/env python3
"""Live API smoke check for parabolic-short-trade-planner.

Five logical checks (4 required gates + 1 optional warning) that prove
both the happy paths and the 404 graceful-handling path that's
otherwise unreachable from Phase 1 output:

    1. FMP historical-price-eod/full?symbol=AAPL&from=&to=  (Issue #64
       flat-list shape; the contract Phase 1 exercises)
    2. FMP profile/AAPL                                     (mktCap)
    3. FMP sp500_constituent                                (optional —
       entitlement varies by tier; print a warning, do not fail)
    4. Alpaca /v2/assets/AAPL                               (shortable +
       easy_to_borrow keys present)
    5. Alpaca /v2/assets/XXXXXFAKE                          (negative
       path: 404 → AlpacaInventoryAdapter must return asset_not_found
       dict without raising)

Note: check #5 issues TWO HTTP requests against the same Alpaca
endpoint (one raw probe to confirm the 404 itself, then a second one
through `AlpacaInventoryAdapter.get_inventory_status` to confirm the
adapter handles that response gracefully). So while there are 5
logical checks, the script makes 6 HTTP calls per run (3 FMP +
3 Alpaca). The cost remains trivial under both API rate limits.

Returns 0 on all four required gates passing. Logs status code +
first 200 chars of the error body on failures.
"""

from __future__ import annotations

import argparse
import os
import sys
from datetime import date, timedelta
from pathlib import Path

import requests

# Ensure the skill's adapters/ is importable for the Alpaca 404 check.
SCRIPTS_DIR = Path(__file__).resolve().parent
ADAPTERS_DIR = SCRIPTS_DIR / "adapters"
for _p in (str(ADAPTERS_DIR), str(SCRIPTS_DIR)):
    if _p not in sys.path:
        sys.path.insert(0, _p)


FMP_STABLE = "https://financialmodelingprep.com/stable"
FMP_STABLE_HIST = "https://financialmodelingprep.com/stable/historical-price-eod/full"
FMP_V3 = "https://financialmodelingprep.com/api/v3"
ALPACA_PAPER = "https://paper-api.alpaca.markets"
ALPACA_LIVE = "https://api.alpaca.markets"

REQUIRED_HISTORICAL_KEYS = {"date", "open", "high", "low", "close", "volume"}


def _truncate(body: str, n: int = 200) -> str:
    body = body.strip()
    return body if len(body) <= n else body[:n] + "..."


def _print_pass(name: str, detail: str) -> None:
    print(f"PASS {name} — {detail}")


def _print_warn(name: str, detail: str) -> None:
    print(f"WARN {name} — {detail}")


def _print_fail(name: str, detail: str) -> None:
    print(f"FAIL {name} — {detail}")


def _print_skip(name: str, detail: str) -> None:
    print(f"SKIP {name} — {detail}")


def check_fmp_historical(api_key: str) -> bool:
    """Gate 1 — verify Issue #64 flat-list shape."""
    name = "fmp.historical_price_eod_full"
    today = date.today()
    params = {
        "symbol": "AAPL",
        "from": (today - timedelta(days=10)).isoformat(),
        "to": today.isoformat(),
        "apikey": api_key,
    }
    try:
        r = requests.get(FMP_STABLE_HIST, params=params, timeout=15)
    except requests.RequestException as e:
        _print_fail(name, f"network error: {e}")
        return False

    if r.status_code != 200:
        _print_fail(name, f"HTTP {r.status_code} — {_truncate(r.text)}")
        return False

    try:
        data = r.json()
    except ValueError:
        _print_fail(name, f"non-JSON body: {_truncate(r.text)}")
        return False

    if not isinstance(data, list) or not data:
        _print_fail(name, f"expected non-empty list[dict], got {type(data).__name__}")
        return False

    first = data[0]
    if not isinstance(first, dict):
        _print_fail(name, f"first row is {type(first).__name__}, expected dict")
        return False

    missing = REQUIRED_HISTORICAL_KEYS - set(first.keys())
    if missing:
        _print_fail(name, f"row missing keys: {sorted(missing)} (got {sorted(first.keys())})")
        return False

    _print_pass(name, f"{len(data)} bars; Issue #64 shape verified (date/o/h/l/c/v)")
    return True


def check_fmp_profile(api_key: str) -> bool:
    """Gate 2 — verify profile returns a non-empty list with a market-cap field.

    Tries /stable/profile?symbol=AAPL first, then the v3 /profile/AAPL fallback
    (matching the client). /stable renamed mktCap -> marketCap, so either is OK.
    """
    name = "fmp.profile"
    attempts = [
        (f"{FMP_STABLE}/profile", {"symbol": "AAPL", "apikey": api_key}),
        (f"{FMP_V3}/profile/AAPL", {"apikey": api_key}),
    ]
    last_detail = "no response"
    for url, params in attempts:
        try:
            r = requests.get(url, params=params, timeout=15)
        except requests.RequestException as e:
            last_detail = f"network error: {e}"
            continue
        if r.status_code != 200:
            last_detail = f"HTTP {r.status_code} — {_truncate(r.text)}"
            continue
        try:
            data = r.json()
        except ValueError:
            last_detail = f"non-JSON body: {_truncate(r.text)}"
            continue
        if isinstance(data, list) and data:
            cap = data[0].get("mktCap", data[0].get("marketCap"))
            if cap is not None:
                _print_pass(name, f"marketCap={cap}")
                return True
        last_detail = f"expected list with market cap on first item, got {data!r:.200}"

    _print_fail(name, last_detail)
    return False


def check_fmp_sp500(api_key: str) -> bool:
    """Optional warning — sp500 constituent entitlement varies by FMP tier.

    Tries /stable/sp500-constituent first, then the v3 /sp500_constituent
    fallback.
    """
    name = "fmp.sp500_constituent"
    attempts = [
        (f"{FMP_STABLE}/sp500-constituent", {"apikey": api_key}),
        (f"{FMP_V3}/sp500_constituent", {"apikey": api_key}),
    ]
    last_detail = "no response"
    for url, params in attempts:
        try:
            r = requests.get(url, params=params, timeout=15)
        except requests.RequestException as e:
            last_detail = f"network error (optional gate): {e}"
            continue
        if r.status_code != 200:
            last_detail = (
                f"HTTP {r.status_code} — likely entitlement (skip on Free); "
                f"body: {_truncate(r.text)}"
            )
            continue
        try:
            data = r.json()
        except ValueError:
            last_detail = f"non-JSON body: {_truncate(r.text)}"
            continue
        if isinstance(data, list) and data:
            _print_pass(name, f"{len(data)} constituents")
            return True
        last_detail = "empty list (skip on Free)"

    _print_warn(name, last_detail)
    return True


def check_alpaca_assets_aapl(api_key: str, secret: str, paper: bool) -> bool:
    """Gate 3 — verify shortable + easy_to_borrow keys are present."""
    name = "alpaca.assets_aapl"
    base = ALPACA_PAPER if paper else ALPACA_LIVE
    url = f"{base}/v2/assets/AAPL"
    headers = {"APCA-API-KEY-ID": api_key, "APCA-API-SECRET-KEY": secret}
    try:
        r = requests.get(url, headers=headers, timeout=15)
    except requests.RequestException as e:
        _print_fail(name, f"network error: {e}")
        return False

    if r.status_code != 200:
        _print_fail(name, f"HTTP {r.status_code} — {_truncate(r.text)}")
        return False

    try:
        data = r.json()
    except ValueError:
        _print_fail(name, f"non-JSON body: {_truncate(r.text)}")
        return False

    missing = {"shortable", "easy_to_borrow"} - set(data.keys())
    if missing:
        _print_fail(name, f"missing keys {sorted(missing)} (got {sorted(data.keys())})")
        return False

    _print_pass(
        name,
        f"shortable={data['shortable']} easy_to_borrow={data['easy_to_borrow']} (paper={paper})",
    )
    return True


def check_alpaca_404_graceful(api_key: str, secret: str, paper: bool) -> bool:
    """Gate 4 — confirm AlpacaInventoryAdapter returns asset_not_found
    on 404 instead of raising. Phase 1 normally rejects unknown tickers
    before they reach the Adapter, so this is the only end-to-end
    coverage of the 404 fix."""
    name = "alpaca.assets_404_graceful"
    base = ALPACA_PAPER if paper else ALPACA_LIVE
    url = f"{base}/v2/assets/XXXXXFAKE"
    headers = {"APCA-API-KEY-ID": api_key, "APCA-API-SECRET-KEY": secret}

    try:
        r = requests.get(url, headers=headers, timeout=15)
    except requests.RequestException as e:
        _print_fail(name, f"network error on raw 404 probe: {e}")
        return False

    if r.status_code != 404:
        _print_fail(
            name,
            f"expected HTTP 404 from raw probe, got {r.status_code} — {_truncate(r.text)}",
        )
        return False

    # Now confirm the Adapter handles the same response gracefully.
    try:
        from alpaca_inventory_adapter import AlpacaInventoryAdapter  # noqa: WPS433

        adapter = AlpacaInventoryAdapter(api_key=api_key, secret_key=secret, paper=paper)
        status = adapter.get_inventory_status("XXXXXFAKE")
    except Exception as e:  # noqa: BLE001 — we explicitly want any raise to fail this gate
        _print_fail(name, f"AlpacaInventoryAdapter raised on 404: {type(e).__name__}: {e}")
        return False

    if status.get("error") != "asset_not_found":
        _print_fail(name, f"expected error=asset_not_found, got {status!r:.200}")
        return False
    if status.get("can_open_new_short") is not False:
        _print_fail(
            name, f"can_open_new_short must be False, got {status.get('can_open_new_short')}"
        )
        return False

    _print_pass(name, "raw HTTP 404 mapped to asset_not_found dict (no exception)")
    return True


def _build_arg_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="Live API smoke check for parabolic-short-trade-planner",
    )
    p.add_argument(
        "--fmp-only",
        action="store_true",
        help=(
            "Skip Alpaca gates entirely. Useful for contributors who only have "
            "an FMP key configured. Exit 0 when the FMP required gates pass; "
            "the Alpaca gates are reported as SKIP and never gate the exit code."
        ),
    )
    return p


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

    print("=" * 70)
    print("Parabolic Short — live API smoke check")
    if args.fmp_only:
        print("Mode: --fmp-only (Alpaca gates will be skipped)")
    print("=" * 70)

    fmp_key = os.environ.get("FMP_API_KEY")
    alpaca_key = os.environ.get("ALPACA_API_KEY")
    alpaca_secret = os.environ.get("ALPACA_SECRET_KEY")
    alpaca_paper = os.environ.get("ALPACA_PAPER", "true").lower() == "true"

    if not fmp_key:
        print("FAIL setup — FMP_API_KEY env var is missing")
        return 1

    # Run the FMP gates first; they don't depend on Alpaca config.
    results: dict[str, bool] = {
        "fmp_historical": check_fmp_historical(fmp_key),
        "fmp_profile": check_fmp_profile(fmp_key),
        "fmp_sp500": check_fmp_sp500(fmp_key),  # optional, never gates exit code
    }

    # Decide whether to run the Alpaca gates.
    alpaca_skipped = args.fmp_only or not alpaca_key or not alpaca_secret

    if alpaca_skipped:
        if args.fmp_only:
            reason = "explicitly skipped via --fmp-only"
        else:
            reason = "ALPACA_API_KEY / ALPACA_SECRET_KEY env vars not configured"
        _print_skip("alpaca.assets_aapl", reason)
        _print_skip("alpaca.assets_404_graceful", reason)
    else:
        results["alpaca_assets_aapl"] = check_alpaca_assets_aapl(
            alpaca_key, alpaca_secret, alpaca_paper
        )
        results["alpaca_404_graceful"] = check_alpaca_404_graceful(
            alpaca_key, alpaca_secret, alpaca_paper
        )

    # Required-gate set depends on whether Alpaca was skipped. In either case
    # the FMP gates are required; the Alpaca gates are only required when
    # they were actually attempted.
    required = ["fmp_historical", "fmp_profile"]
    if not alpaca_skipped:
        required.extend(["alpaca_assets_aapl", "alpaca_404_graceful"])

    passed = sum(1 for k in required if results.get(k))
    print("-" * 70)
    if alpaca_skipped:
        print(
            f"Required gates: {passed}/{len(required)} passed "
            f"(FMP only; Alpaca gates skipped, sp500 is optional warning)"
        )
        if not args.fmp_only:
            print(
                "Note: pass --fmp-only to acknowledge intent, or set ALPACA_API_KEY / "
                "ALPACA_SECRET_KEY to run the full check."
            )
    else:
        print(f"Required gates: {passed}/{len(required)} passed (sp500 is optional warning)")

    return 0 if passed == len(required) else 1


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

#!/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 Parabolic Short Trade Planner

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
- Batch quote support
- S&P 500 constituents 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_quote_url(base, symbols_str, params):
    """stable/quote?symbol=^GSPC"""
    params["symbol"] = symbols_str
    return base, params


def _v3_quote_url(base, symbols_str, params):
    """api/v3/quote/^GSPC"""
    return f"{base}/{symbols_str}", params


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 = {
    "quote": [
        ("https://financialmodelingprep.com/stable/quote", _stable_quote_url),
        ("https://financialmodelingprep.com/api/v3/quote", _v3_quote_url),
    ],
    "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

    def __init__(self, api_key: Optional[str] = None):
        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
        # 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]:
        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 == "quote":
                if not isinstance(data, list) or len(data) == 0:
                    valid = False
                    shape_issue = "expected non-empty list"
                elif is_single and not any(
                    q.get("symbol", "").replace("-", ".") == symbols_str.replace("-", ".")
                    for q in data
                ):
                    valid = False
                    shape_issue = f"requested symbol '{symbols_str}' not in response"

            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_sp500_constituents(self) -> Optional[list[dict]]:
        """Fetch S&P 500 constituent list.

        Returns:
            List of dicts with keys: symbol, name, sector, subSector
            or None on failure.
        """
        cache_key = "sp500_constituents"
        if cache_key in self.cache:
            return self.cache[cache_key]

        # Migrate hardcoded v3 URL to /stable (this method bypasses the
        # _FMP_ENDPOINTS stable→v3 fallback list, so rewrite at the call site).
        url, params = v3_to_stable(f"{self.BASE_URL}/sp500_constituent")
        data = self._rate_limited_get(url, params)
        if data:
            self.cache[cache_key] = data
        return data

    def get_quote(self, symbols: str) -> Optional[list[dict]]:
        """Fetch real-time quote data for one or more symbols (comma-separated)"""
        cache_key = f"quote_{symbols}"
        if cache_key in self.cache:
            return self.cache[cache_key]

        data = self._request_with_fallback("quote", symbols)
        if data:
            self.cache[cache_key] = data
        return data

    def get_batch_quotes(self, symbols: list[str]) -> dict[str, dict]:
        """Fetch quotes for a list of symbols, batching up to 5 per request"""
        results = {}
        batch_size = 5
        for i in range(0, len(symbols), batch_size):
            batch = symbols[i : i + batch_size]
            batch_str = ",".join(batch)
            quotes = self.get_quote(batch_str)
            if quotes:
                for q in quotes:
                    results[q["symbol"]] = q
        return results

    def get_batch_historical(self, symbols: list[str], days: int = 260) -> dict[str, list[dict]]:
        """Fetch historical prices for multiple symbols"""
        results = {}
        for symbol in symbols:
            data = self.get_historical_prices(symbol, days=days)
            if data and "historical" in data:
                results[symbol] = data["historical"]
        return results

    def calculate_sma(self, prices: list[float], period: int) -> float:
        """Calculate Simple Moving Average from a list of prices (most recent first)"""
        if len(prices) < period:
            return sum(prices) / len(prices)
        return sum(prices[:period]) / period

    def get_company_profile(self, symbol: str) -> Optional[dict]:
        """Fetch company profile (market cap, IPO date, sector, etc.).

        Uses ``/stable/profile?symbol={symbol}`` (single symbol). Returns
        ``None`` on failure.
        """
        cache_key = f"profile_{symbol}"
        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}/profile/{symbol}")
        data = self._rate_limited_get(url, params)
        if isinstance(data, list) and data:
            profile = data[0]
            # /stable/profile returns ``marketCap``; v3 returned ``mktCap``.
            # Re-alias so consumers (e.g. screen_parabolic.py market-cap bounds)
            # keep working against the legacy key.
            if "mktCap" not in profile and "marketCap" in profile:
                profile["mktCap"] = profile["marketCap"]
            self.cache[cache_key] = profile
            return profile
        return None

    def get_profile_bulk(self, part: int = 0) -> Optional[list[dict]]:
        """Bulk profile download (Premium endpoint).

        Used to avoid per-symbol calls during the initial universe pass.
        """
        cache_key = f"profile_bulk_{part}"
        if cache_key in self.cache:
            return self.cache[cache_key]
        url = "https://financialmodelingprep.com/stable/profile-bulk"
        data = self._rate_limited_get(url, params={"part": part}, quiet=True)
        if isinstance(data, list) and data:
            self.cache[cache_key] = data
            return data
        return None

    def get_earnings_calendar(self, from_date: str, to_date: str) -> Optional[list[dict]]:
        """Fetch upcoming earnings between two dates (YYYY-MM-DD)."""
        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=params)
        if isinstance(data, list):
            self.cache[cache_key] = data
        return data

    def get_aftermarket_quote(self, symbol: str) -> Optional[dict]:
        """Pre/after-market quote.

        FMP exposes this as ``stable/aftermarket-quote?symbol=AAPL``. The
        response shape is ``price`` / ``bid`` / ``ask`` / ``volume`` /
        ``timestamp`` — ``high`` / ``low`` are NOT guaranteed and the caller
        must handle ``None`` for those fields explicitly. The returned dict
        always carries a ``source`` key for the schema.

        Returns ``None`` on failure (no aftermarket data published).
        """
        cache_key = f"aftermarket_{symbol}"
        if cache_key in self.cache:
            return self.cache[cache_key]
        url = "https://financialmodelingprep.com/stable/aftermarket-quote"
        data = self._rate_limited_get(url, params={"symbol": symbol}, quiet=True)
        if not data:
            return None
        # FMP may return either a list or a dict depending on tier
        row = data[0] if isinstance(data, list) and data else data
        if not isinstance(row, dict):
            return None
        out = {
            "price": row.get("price"),
            "bid": row.get("bid"),
            "ask": row.get("ask"),
            "volume": row.get("volume") or row.get("size"),
            "high": row.get("high"),  # may be None
            "low": row.get("low"),  # may be None
            "timestamp": row.get("timestamp") or row.get("date"),
            "source": "fmp_aftermarket_quote",
        }
        self.cache[cache_key] = out
        return out

    def get_intraday_ohlcv(
        self,
        symbol: str,
        interval: str = "5min",
        from_ts: Optional[str] = None,
        to_ts: Optional[str] = None,
    ) -> Optional[list[dict]]:
        """Intraday bars — Phase 3 (v0.5) feature.

        Stub for the MVP. Phase 1 + Phase 2 do not need this; the
        function is declared so the broader skill API surface is stable.
        """
        raise NotImplementedError(
            "Intraday OHLCV is part of Phase 3 (v0.5). MVP only consumes "
            "daily bars + aftermarket quote."
        )

    def get_historical_prices(self, symbol: str, days: int = 365) -> 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,
            "rate_limit_reached": self.rate_limit_reached,
        }