flashalpha-python/src/flashalpha/client.py at main · FlashAlpha-lab/flashalpha-python

707 lines (615 loc) · 27.7 KB
"""FlashAlpha API client."""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from urllib.parse import quote
import requests
from .exceptions import (
    AuthenticationError,
    FlashAlphaError,
    NotFoundError,
    RateLimitError,
    ServerError,
    TierRestrictedError,
if TYPE_CHECKING:
    from .types import (
        AccountResponse,
        AdvVolatilityResponse,
        ChexResponse,
        DexResponse,
        ExposureLevelsResponse,
        ExposureSummaryResponse,
        FlowDealerRiskResponse,
        FlowDexResponse,
        FlowGexResponse,
        FlowLevelsResponse,
        FlowLiveResponse,
        FlowOiResponse,
        FlowOptionBlocksResponse,
        FlowOptionCumulativeResponse,
        FlowOptionHistoryResponse,
        FlowOptionLeaderboardResponse,
        FlowOptionOutliersResponse,
        FlowOptionRecentResponse,
        FlowOptionSummaryResponse,
        FlowPinRiskResponse,
        FlowStockBlocksResponse,
        FlowStockCumulativeResponse,
        FlowStockHistoryResponse,
        FlowStockLeaderboardResponse,
        FlowStockOutliersResponse,
        FlowStockRecentResponse,
        FlowStockSummaryResponse,
        FlowSummaryResponse,
        GexResponse,
        HealthResponse,
        MaxPainResponse,
        NarrativeResponse,
        OptionQuoteResponse,
        OptionsMetaResponse,
        PricingGreeksResponse,
        PricingIvResponse,
        PricingKellyResponse,
        ScreenerResponse,
        StockQuoteResponse,
        StockSummaryResponse,
        SurfaceResponse,
        SymbolsResponse,
        TickersResponse,
        VexResponse,
        VolatilityResponse,
        VrpResponse,
        ZeroDteResponse,
BASE_URL = "https://lab.flashalpha.com"
def _seg(s: str) -> str:
    """URL-escape a single path segment (e.g. a ticker) — escapes / ? % etc."""
    return quote(s, safe="")
class FlashAlpha:
    """Thin wrapper around the FlashAlpha REST API.
    Parameters
    ----------
    api_key : str
        Your FlashAlpha API key from https://flashalpha.com
    base_url : str, optional
        Override the API base URL (for testing).
    timeout : float, optional
        Request timeout in seconds. Default 30.
    def __init__(self, api_key: str, *, base_url: str = BASE_URL, timeout: float = 30):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self._session = requests.Session()
        self._session.headers["X-Api-Key"] = api_key
    # ── internal ────────────────────────────────────────────────────
    def _get(self, path: str, params: dict[str, Any] | None = None) -> dict:
        url = f"{self.base_url}{path}"
        resp = self._session.get(url, params=params, timeout=self.timeout)
        return self._handle(resp)
    def _post(self, path: str, json_body: dict[str, Any] | None = None) -> dict:
        url = f"{self.base_url}{path}"
        resp = self._session.post(url, json=json_body, timeout=self.timeout)
        return self._handle(resp)
    def _handle(self, resp: requests.Response) -> dict:
        if resp.status_code == 200:
            return resp.json()
        try:
            body = resp.json()
        except ValueError:
            body = {"detail": resp.text}
        msg = body.get("message") or body.get("detail") or resp.text
        if resp.status_code == 401:
            raise AuthenticationError(msg, status_code=401, response=body)
        if resp.status_code == 403:
            raise TierRestrictedError(
                status_code=403,
                response=body,
                current_plan=body.get("current_plan"),
                required_plan=body.get("required_plan"),
        if resp.status_code == 404:
            raise NotFoundError(msg, status_code=404, response=body)
        if resp.status_code == 429:
            raise RateLimitError(
                status_code=429,
                response=body,
                retry_after=int(resp.headers.get("Retry-After", 0)) or None,
        if resp.status_code >= 500:
            raise ServerError(msg, status_code=resp.status_code, response=body)
        raise FlashAlphaError(msg, status_code=resp.status_code, response=body)
    # ── Market Data ─────────────────────────────────────────────────
    def stock_quote(self, ticker: str) -> StockQuoteResponse:
        """Live stock quote (bid/ask/mid/last)."""
        return self._get(f"/stockquote/{_seg(ticker)}")
    def option_quote(
        self,
        ticker: str,
        expiry: str | None = None,
        strike: float | None = None,
        type: str | None = None,
    ) -> OptionQuoteResponse | list[OptionQuoteResponse]:
        """Option quotes with greeks. Requires Growth+.
        Returns a single ``OptionQuoteResponse`` when the request fully
        specifies one contract (all of ``expiry``, ``strike``, ``type``);
        otherwise returns a list of ``OptionQuoteResponse``.
        """
        params: dict[str, Any] = {}
        if expiry:
            params["expiry"] = expiry
        if strike is not None:
            params["strike"] = strike
        if type:
            params["type"] = type
        return self._get(f"/optionquote/{_seg(ticker)}", params or None)
    def surface(self, symbol: str) -> SurfaceResponse:
        """Volatility surface grid (public, no auth required)."""
        return self._get(f"/v1/surface/{_seg(symbol)}")
    def stock_summary(self, symbol: str) -> StockSummaryResponse:
        """Comprehensive stock summary (price, vol, exposure, macro)."""
        return self._get(f"/v1/stock/{_seg(symbol)}/summary")
    # ── Historical ──────────────────────────────────────────────────
    def historical_stock_quote(self, ticker: str, *, date: str, time: str | None = None) -> dict:
        """Historical stock quotes (minute-by-minute from ClickHouse)."""
        params: dict[str, Any] = {"date": date}
        if time:
            params["time"] = time
        return self._get(f"/historical/stockquote/{_seg(ticker)}", params)
    def historical_option_quote(
        self,
        ticker: str,
        date: str,
        time: str | None = None,
        expiry: str | None = None,
        strike: float | None = None,
        type: str | None = None,
    ) -> dict:
        """Historical option quotes (minute-by-minute from ClickHouse)."""
        params: dict[str, Any] = {"date": date}
        if time:
            params["time"] = time
        if expiry:
            params["expiry"] = expiry
        if strike is not None:
            params["strike"] = strike
        if type:
            params["type"] = type
        return self._get(f"/historical/optionquote/{_seg(ticker)}", params)
    # ── Exposure Analytics ──────────────────────────────────────────
    def gex(self, symbol: str, *, expiration: str | None = None, min_oi: int | None = None) -> GexResponse:
        """Gamma exposure by strike."""
        params: dict[str, Any] = {}
        if expiration:
            params["expiration"] = expiration
        if min_oi is not None:
            params["min_oi"] = min_oi
        return self._get(f"/v1/exposure/gex/{_seg(symbol)}", params or None)
    def dex(self, symbol: str, *, expiration: str | None = None) -> DexResponse:
        """Delta exposure by strike."""
        params: dict[str, Any] = {}
        if expiration:
            params["expiration"] = expiration
        return self._get(f"/v1/exposure/dex/{_seg(symbol)}", params or None)
    def vex(self, symbol: str, *, expiration: str | None = None) -> VexResponse:
        """Vanna exposure by strike."""
        params: dict[str, Any] = {}
        if expiration:
            params["expiration"] = expiration
        return self._get(f"/v1/exposure/vex/{_seg(symbol)}", params or None)
    def chex(self, symbol: str, *, expiration: str | None = None) -> ChexResponse:
        """Charm exposure by strike."""
        params: dict[str, Any] = {}
        if expiration:
            params["expiration"] = expiration
        return self._get(f"/v1/exposure/chex/{_seg(symbol)}", params or None)
    def exposure_summary(self, symbol: str) -> ExposureSummaryResponse:
        """Full exposure summary (GEX/DEX/VEX/CHEX + hedging). Requires Growth+."""
        return self._get(f"/v1/exposure/summary/{_seg(symbol)}")
    def exposure_levels(self, symbol: str) -> ExposureLevelsResponse:
        """Key support/resistance levels from options exposure."""
        return self._get(f"/v1/exposure/levels/{_seg(symbol)}")
    def narrative(self, symbol: str) -> NarrativeResponse:
        """Verbal narrative analysis of exposure. Requires Growth+."""
        return self._get(f"/v1/exposure/narrative/{_seg(symbol)}")
    def zero_dte(self, symbol: str, *, strike_range: float | None = None) -> ZeroDteResponse:
        """Real-time 0DTE analytics: regime, expected move, pin risk, hedging, decay. Requires Growth+.
        Returns a ``ZeroDteResponse`` (a ``TypedDict`` — runtime-equivalent to
        ``dict``). Existing ``result["field"]`` access continues to work; new
        callers get autocomplete and type-checking on the documented fields.
        """
        params: dict[str, Any] = {}
        if strike_range is not None:
            params["strike_range"] = strike_range
        return self._get(f"/v1/exposure/zero-dte/{_seg(symbol)}", params or None)
    # ── Flow (live, simulation-aware) — requires the Alpha plan ──────
    # Analytics endpoints (snake_case wire shape) fold today's intraday
    # trade tape into the settled book. All accept an optional
    # ``expiry="YYYY-MM-DD"`` to slice to a single expiration cycle.
    def flow_levels(self, symbol: str, *, expiry: str | None = None) -> FlowLevelsResponse:
        """Live gamma flip / call & put walls / max pain. Requires Alpha."""
        params: dict[str, Any] = {}
        if expiry:
            params["expiry"] = expiry
        return self._get(f"/v1/flow/levels/{_seg(symbol)}", params or None)
    def flow_pin_risk(self, symbol: str, *, expiry: str | None = None) -> FlowPinRiskResponse:
        """0DTE pin-risk score + component breakdown. Requires Alpha."""
        params: dict[str, Any] = {}
        if expiry:
            params["expiry"] = expiry
        return self._get(f"/v1/flow/pin-risk/{_seg(symbol)}", params or None)
    def flow_summary(self, symbol: str, *, expiry: str | None = None) -> FlowSummaryResponse:
        """At-a-glance flow direction + headline GEX shift. Requires Alpha."""
        params: dict[str, Any] = {}
        if expiry:
            params["expiry"] = expiry
        return self._get(f"/v1/flow/summary/{_seg(symbol)}", params or None)
    def flow_oi(self, symbol: str, *, expiry: str | None = None) -> FlowOiResponse:
        """Open-interest simulator state (official vs intraday). Requires Alpha."""
        params: dict[str, Any] = {}
        if expiry:
            params["expiry"] = expiry
        return self._get(f"/v1/flow/oi/{_seg(symbol)}", params or None)
    def flow_gex(self, symbol: str, *, expiry: str | None = None) -> FlowGexResponse:
        """Live (flow-adjusted) GEX + per-strike profile. Requires Alpha."""
        params: dict[str, Any] = {}
        if expiry:
            params["expiry"] = expiry
        return self._get(f"/v1/flow/gex/{_seg(symbol)}", params or None)
    def flow_dex(self, symbol: str, *, expiry: str | None = None) -> FlowDexResponse:
        """Live (flow-adjusted) DEX + per-strike profile. Requires Alpha."""
        params: dict[str, Any] = {}
        if expiry:
            params["expiry"] = expiry
        return self._get(f"/v1/flow/dex/{_seg(symbol)}", params or None)
    def flow_dealer_risk(self, symbol: str, *, expiry: str | None = None) -> FlowDealerRiskResponse:
        """Settled-vs-live dealer GEX/DEX + flow adjustment. Requires Alpha."""
        params: dict[str, Any] = {}
        if expiry:
            params["expiry"] = expiry
        return self._get(f"/v1/flow/dealer-risk/{_seg(symbol)}", params or None)
    def flow_live(self, symbol: str, *, expiry: str | None = None) -> FlowLiveResponse:
        """Everything-at-once live flow bundle (convenience). Requires Alpha."""
        params: dict[str, Any] = {}
        if expiry:
            params["expiry"] = expiry
        return self._get(f"/v1/flow/live/{_seg(symbol)}", params or None)
    # Raw flow data (camelCase wire shape) — proxied trade tape.
    def flow_option_recent(
        self, symbol: str, *, limit: int | None = None, expiry: str | None = None
    ) -> FlowOptionRecentResponse:
        """Recent option trades, newest-first (``limit`` 1–500). Requires Alpha."""
        params: dict[str, Any] = {}
        if limit is not None:
            params["limit"] = limit
        if expiry:
            params["expiry"] = expiry
        return self._get(f"/v1/flow/options/{_seg(symbol)}/recent", params or None)
    def flow_option_summary(
        self, symbol: str, *, expiry: str | None = None
    ) -> FlowOptionSummaryResponse:
        """Per-underlying option-flow aggregates. Requires Alpha."""
        params: dict[str, Any] = {}
        if expiry:
            params["expiry"] = expiry
        return self._get(f"/v1/flow/options/{_seg(symbol)}/summary", params or None)
    def flow_option_blocks(
        self, symbol: str, *, min_size: int | None = None, expiry: str | None = None
    ) -> FlowOptionBlocksResponse:
        """Large option prints (``size >= min_size``). Requires Alpha."""
        params: dict[str, Any] = {}
        if min_size is not None:
            params["minSize"] = min_size
        if expiry:
            params["expiry"] = expiry
        return self._get(f"/v1/flow/options/{_seg(symbol)}/blocks", params or None)
    def flow_option_history(
        self, symbol: str, *, minutes: int | None = None, expiry: str | None = None
    ) -> FlowOptionHistoryResponse:
        """Per-minute option-flow buckets (``minutes`` 1–10080). Requires Alpha."""
        params: dict[str, Any] = {}
        if minutes is not None:
            params["minutes"] = minutes
        if expiry:
            params["expiry"] = expiry
        return self._get(f"/v1/flow/options/{_seg(symbol)}/history", params or None)
    def flow_option_cumulative(
        self, symbol: str, *, minutes: int | None = None, expiry: str | None = None
    ) -> FlowOptionCumulativeResponse:
        """Cumulative option net-flow series. Requires Alpha."""
        params: dict[str, Any] = {}
        if minutes is not None:
            params["minutes"] = minutes
        if expiry:
            params["expiry"] = expiry
        return self._get(f"/v1/flow/options/{_seg(symbol)}/cumulative", params or None)
    def flow_stock_recent(
        self, symbol: str, *, limit: int | None = None
    ) -> FlowStockRecentResponse:
        """Recent stock trades, newest-first (``limit`` 1–500). Requires Alpha."""
        params: dict[str, Any] = {}
        if limit is not None:
            params["limit"] = limit
        return self._get(f"/v1/flow/stocks/{_seg(symbol)}/recent", params or None)
    def flow_stock_summary(self, symbol: str) -> FlowStockSummaryResponse:
        """Per-symbol stock-flow aggregates. Requires Alpha."""
        return self._get(f"/v1/flow/stocks/{_seg(symbol)}/summary")
    def flow_stock_blocks(
        self, symbol: str, *, min_size: int | None = None
    ) -> FlowStockBlocksResponse:
        """Large stock prints (``size >= min_size``). Requires Alpha."""
        params: dict[str, Any] = {}
        if min_size is not None:
            params["minSize"] = min_size
        return self._get(f"/v1/flow/stocks/{_seg(symbol)}/blocks", params or None)
    def flow_stock_history(
        self, symbol: str, *, minutes: int | None = None
    ) -> FlowStockHistoryResponse:
        """Per-minute stock-flow buckets w/ OHLC (``minutes`` 1–10080). Requires Alpha."""
        params: dict[str, Any] = {}
        if minutes is not None:
            params["minutes"] = minutes
        return self._get(f"/v1/flow/stocks/{_seg(symbol)}/history", params or None)
    def flow_stock_cumulative(
        self, symbol: str, *, minutes: int | None = None
    ) -> FlowStockCumulativeResponse:
        """Cumulative stock net-flow series. Requires Alpha."""
        params: dict[str, Any] = {}
        if minutes is not None:
            params["minutes"] = minutes
        return self._get(f"/v1/flow/stocks/{_seg(symbol)}/cumulative", params or None)
    def flow_options_leaderboard(
        self, *, n: int | None = None, window_minutes: int | None = None
    ) -> FlowOptionLeaderboardResponse:
        """Cross-symbol option-flow leaderboard (top ``n`` by net $). Requires Alpha."""
        params: dict[str, Any] = {}
        if n is not None:
            params["n"] = n
        if window_minutes is not None:
            params["windowMinutes"] = window_minutes
        return self._get("/v1/flow/options/leaderboard", params or None)
    def flow_options_outliers(
        self,
        limit: int | None = None,
        min_trades: int | None = None,
        window_minutes: int | None = None,
    ) -> FlowOptionOutliersResponse:
        """Cross-symbol option-flow outliers (imbalance-ranked). Requires Alpha."""
        params: dict[str, Any] = {}
        if limit is not None:
            params["limit"] = limit
        if min_trades is not None:
            params["minTrades"] = min_trades
        if window_minutes is not None:
            params["windowMinutes"] = window_minutes
        return self._get("/v1/flow/options/outliers", params or None)
    def flow_stocks_leaderboard(
        self, *, n: int | None = None, window_minutes: int | None = None
    ) -> FlowStockLeaderboardResponse:
        """Cross-symbol stock-flow leaderboard (top ``n`` by net $). Requires Alpha."""
        params: dict[str, Any] = {}
        if n is not None:
            params["n"] = n
        if window_minutes is not None:
            params["windowMinutes"] = window_minutes
        return self._get("/v1/flow/stocks/leaderboard", params or None)
    def flow_stocks_outliers(
        self,
        limit: int | None = None,
        min_trades: int | None = None,
        window_minutes: int | None = None,
    ) -> FlowStockOutliersResponse:
        """Cross-symbol stock-flow outliers (imbalance-ranked). Requires Alpha."""
        params: dict[str, Any] = {}
        if limit is not None:
            params["limit"] = limit
        if min_trades is not None:
            params["minTrades"] = min_trades
        if window_minutes is not None:
            params["windowMinutes"] = window_minutes
        return self._get("/v1/flow/stocks/outliers", params or None)
    # ── Pricing & Sizing ────────────────────────────────────────────
    def greeks(
        self,
        spot: float,
        strike: float,
        dte: float,
        sigma: float,
        type: str = "call",
        r: float | None = None,
        q: float | None = None,
    ) -> PricingGreeksResponse:
        """Full BSM greeks (first, second, third order)."""
        params: dict[str, Any] = {"spot": spot, "strike": strike, "dte": dte, "sigma": sigma, "type": type}
        if r is not None:
            params["r"] = r
        if q is not None:
            params["q"] = q
        return self._get("/v1/pricing/greeks", params)
    def iv(
        self,
        spot: float,
        strike: float,
        dte: float,
        price: float,
        type: str = "call",
        r: float | None = None,
        q: float | None = None,
    ) -> PricingIvResponse:
        """Implied volatility from market price."""
        params: dict[str, Any] = {"spot": spot, "strike": strike, "dte": dte, "price": price, "type": type}
        if r is not None:
            params["r"] = r
        if q is not None:
            params["q"] = q
        return self._get("/v1/pricing/iv", params)
    def kelly(
        self,
        spot: float,
        strike: float,
        dte: float,
        sigma: float,
        premium: float,
        mu: float,
        type: str = "call",
        r: float | None = None,
        q: float | None = None,
    ) -> PricingKellyResponse:
        """Kelly criterion optimal position sizing. Requires Growth+."""
        params: dict[str, Any] = {
            "spot": spot,
            "strike": strike,
            "dte": dte,
            "sigma": sigma,
            "premium": premium,
            "mu": mu,
            "type": type,
        if r is not None:
            params["r"] = r
        if q is not None:
            params["q"] = q
        return self._get("/v1/pricing/kelly", params)
    # ── Volatility Analytics ────────────────────────────────────────
    def volatility(self, symbol: str) -> VolatilityResponse:
        """Comprehensive volatility analysis. Requires Growth+."""
        return self._get(f"/v1/volatility/{_seg(symbol)}")
    def adv_volatility(self, symbol: str) -> AdvVolatilityResponse:
        """Advanced volatility analytics: SVI parameters, variance surface, arbitrage detection, greeks surfaces, variance swap. Requires Alpha+."""
        return self._get(f"/v1/adv_volatility/{_seg(symbol)}")
    # ── Reference Data ──────────────────────────────────────────────
    def tickers(self) -> TickersResponse:
        """All available stock tickers."""
        return self._get("/v1/tickers")
    def options(self, ticker: str) -> OptionsMetaResponse:
        """Option chain metadata (expirations + strikes)."""
        return self._get(f"/v1/options/{_seg(ticker)}")
    def symbols(self) -> SymbolsResponse:
        """Currently queried symbols with live data."""
        return self._get("/v1/symbols")
    # ── VRP (Variance Risk Premium) ─────────────────────────────────
    def vrp(self, symbol: str) -> VrpResponse:
        """Variance risk premium analytics — the implied-vs-realized vol
        spread, conditioned on dealer gamma and vanna regime, with
        strategy scores for harvesting.
        Returns a nested payload. Key access paths:
        - ``response["symbol"]``, ``response["underlying_price"]`` — top-level
        - ``response["vrp"]["z_score"]``, ``["percentile"]``,
          ``["atm_iv"]``, ``["rv_20d"]``, ``["vrp_20d"]`` — core VRP metrics
        - ``response["directional"]["downside_vrp"]``,
          ``["upside_vrp"]`` — directional skew (NOT ``put_vrp``/``call_vrp``)
        - ``response["gex_conditioned"]["harvest_score"]``,
          ``["regime"]`` — gamma-regime conditioning
        - ``response["regime"]["net_gex"]``, ``["gamma"]``,
          ``["vrp_regime"]`` — regime snapshot
        - ``response["strategy_scores"]`` — short_put_spread, short_strangle,
          iron_condor, calendar_spread (0–100)
        - ``response["net_harvest_score"]``,
          ``response["dealer_flow_risk"]`` — top-level composite scores
        Requires Alpha+.
        """
        return self._get(f"/v1/vrp/{_seg(symbol)}")
    # ── Max Pain ────────────────────────────────────────────────────
    def max_pain(self, symbol: str, *, expiration: str | None = None) -> MaxPainResponse:
        """Max pain analysis with dealer alignment overlay, pain curve, OI
        breakdown, expected move context, pin probability, and multi-expiry
        calendar. Requires Growth+.
        Parameters
        ----------
        symbol : str
            Underlying symbol.
        expiration : str, optional
            Filter to single expiry (YYYY-MM-DD). Omit for full-chain analysis.
        """
        params: dict[str, Any] = {}
        if expiration:
            params["expiration"] = expiration
        return self._get(f"/v1/maxpain/{_seg(symbol)}", params or None)
    # ── Screener ────────────────────────────────────────────────────
    def screener(
        self,
        filters: dict[str, Any] | None = None,
        sort: list[dict[str, Any]] | None = None,
        select: list[str] | None = None,
        formulas: list[dict[str, str]] | None = None,
        limit: int | None = None,
        offset: int | None = None,
    ) -> ScreenerResponse:
        """Live options screener — filter/rank symbols by gamma exposure, VRP,
        volatility, greeks, and more.
        Powered by an in-memory store updated every 5-10s from live market data.
        Growth: 10-symbol universe, up to 10 rows. Alpha: ~250 symbols, up to 50
        rows, formulas, and harvest/dealer-flow-risk scores.
        Parameters
        ----------
        filters : dict, optional
            Recursive filter tree. Leaf: {"field": "atm_iv", "operator": "gte",
            "value": 20}. Group: {"op": "and", "conditions": [...]}. Supports
            dotted prefixes `expiries.X`, `strikes.X`, `contracts.X` for
            cascading filters.
        sort : list of dict, optional
            Sort specs (primary first), e.g.
            [{"field": "harvest_score", "direction": "desc"}].
            Also accepts {"formula": "alias_name", ...}.
        select : list of str, optional
            Field names to return, or ["*"] for the full flat object.
        formulas : list of dict, optional
            Computed fields (Alpha only), e.g.
            [{"alias": "vrp_ratio", "expression": "atm_iv / rv_20d"}].
        limit : int, optional
            Row cap. 1-10 on Growth, 1-50 on Alpha. Default 50.
        offset : int, optional
            Pagination offset (Alpha only).
        Returns
        -------
        dict
            {"meta": {"total_count": ..., "tier": ..., ...},
             "data": [{"symbol": ..., ...}, ...]}
        Examples
        --------
        Harvestable VRP screen:
        >>> fa.screener(
        ...     filters={
        ...         "op": "and",
        ...         "conditions": [
        ...             {"field": "regime", "operator": "eq", "value": "positive_gamma"},
        ...             {"field": "vrp_regime", "operator": "eq", "value": "harvestable"},
        ...             {"field": "harvest_score", "operator": "gte", "value": 65},
        ...         ],
        ...     },
        ...     sort=[{"field": "harvest_score", "direction": "desc"}],
        ...     select=["symbol", "price", "harvest_score", "dealer_flow_risk"],
        ... )
        """
        body: dict[str, Any] = {}
        if filters is not None:
            body["filters"] = filters
        if sort is not None:
            body["sort"] = sort
        if select is not None:
            body["select"] = select
        if formulas is not None:
            body["formulas"] = formulas
        if limit is not None:
            body["limit"] = limit
        if offset is not None:
            body["offset"] = offset
        return self._post("/v1/screener", body)
    # ── Account & System ────────────────────────────────────────────
    def account(self) -> AccountResponse:
        """Account info and quota."""
        return self._get("/v1/account")
    def health(self) -> HealthResponse:
        """Health check (public, no auth required)."""
        return self._get("/health")
Provide feedback

Saved searches

Use saved searches to filter your results more quickly

FilesExpand file tree

client.py

Latest commit

History

client.py

File metadata and controls
