chart/scripts/check_schwab.py

"""Find out what a Schwab app is actually entitled to, before building on it.

Answers the three questions that decide the design, empirically rather than
from documentation:

  1. Do the credentials authenticate at all?
  2. Do REST quotes work for /ES — futures market data is a separate
     entitlement from equities and may not be granted.
  3. Does the streamer connect? Its bootstrap reads /trader/v1/userPreference,
     which belongs to the Accounts and Trading product, so an app registered
     for Market Data Production alone is expected to fail there.

Two steps, neither of them interactive, so this works over a pipe or from an
agent session where stdin is not a terminal:

    python3 -m scripts.check_schwab
        prints the Schwab login URL.

    python3 -m scripts.check_schwab --redirect-url 'https://.../api/qt?code=…'
        exchanges the code, saves the token, runs the checks.

The browser does not have to be on this machine. Nothing is captured locally —
you copy a URL out, and paste a URL back. The authorisation code is single use
and expires within minutes, so do not leave it sitting between the two steps.

Once a token exists, running with no arguments skips straight to the checks.
"""
import argparse
import sys
from urllib.parse import parse_qs, urlparse

from app.config import Settings


def heading(text: str) -> None:
    print(f"\n{text}\n{'-' * len(text)}")


def load_settings() -> Settings:
    settings = Settings()
    if not settings.schwab_api_key or not settings.schwab_app_secret:
        raise SystemExit("Set SCHWAB_API_KEY and SCHWAB_APP_SECRET in .env first")
    return settings


def key_is_live(authorization_url: str) -> bool:
    """Ask Schwab whether it recognises the app key, before opening a browser.

    A key Schwab does not know produces `invalid_client` here — and so does a
    deliberately invented one, byte for byte, so this cannot tell "wrong value"
    from "not active yet". It can still save a confusing round trip through the
    login page.
    """
    import httpx

    try:
        response = httpx.get(authorization_url, follow_redirects=False, timeout=15)
    except Exception:
        return True  # Network trouble is not evidence about the key.
    return "invalid_client" not in response.text


def secret_is_valid(settings: Settings) -> bool | None:
    """Check the secret without needing an authorisation code.

    The token endpoint authenticates the key and secret over HTTP Basic before
    it looks at the grant, so a deliberately invalid code separates the two
    failures: bad credentials give invalid_client, good credentials give
    invalid_grant. Otherwise a truncated secret survives the login unnoticed and
    only surfaces at the exchange, after the code has been spent.

    None when the answer is not clear enough to act on.
    """
    import httpx

    try:
        response = httpx.post(
            "https://api.schwabapi.com/v1/oauth/token",
            auth=(settings.schwab_api_key, settings.schwab_app_secret),
            data={
                "grant_type": "authorization_code",
                "code": "deliberately-invalid-code",
                "redirect_uri": settings.schwab_callback_url,
            },
            timeout=20,
        )
    except Exception:
        return None
    if "invalid_grant" in response.text:
        return True
    if "invalid_client" in response.text or response.status_code == 401:
        return False
    return None


def print_login_url(settings: Settings) -> None:
    from schwab.auth import get_auth_context

    context = get_auth_context(settings.schwab_api_key, settings.schwab_callback_url)

    if secret_is_valid(settings) is False:
        print("Schwab rejects the app key and secret pair.\n")
        print("The key alone is accepted at the authorize endpoint, so this is")
        print("the secret. Re-copy it from the portal using the Show icon —")
        print("a value clipped by one character looks entirely normal.")
        sys.exit(1)

    if not key_is_live(context.authorization_url):
        print("Schwab rejects this app key with invalid_client.\n")
        print("The secret is not involved yet — it is only used when the code is")
        print("exchanged — so this is the key itself or the app's readiness.\n")
        print("  1. Re-copy the App Key from the portal using the Show icon.")
        print("  2. If it matches, the app is most likely not live yet. Newly")
        print("     created or newly edited apps take time to propagate, and the")
        print("     portal says Ready For Use before the key works.")
        print("\nRe-run this to check again; nothing else is needed.")
        sys.exit(1)

    print("Open this in any browser, on any machine, and approve the app:\n")
    print(f"  {context.authorization_url}\n")
    print("You will land on the callback URL. A 404 there is fine until the")
    print("branch is deployed — the code is in the address bar either way.")
    print("Copy the ENTIRE address and run:\n")
    print("  python3 -m scripts.check_schwab --redirect-url '<paste it here>'")


def ensure_token_path_writable(settings: Settings) -> None:
    path = settings.schwab_token_path
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        probe = path.parent / f".{path.name}.probe"
        probe.touch()
        probe.unlink()
    except OSError as error:
        raise SystemExit(
            f"Cannot write the token to {path} ({error.strerror}).\n"
            f"Point SCHWAB_TOKEN_PATH at a directory you own and try again — "
            f"the authorisation code is spent either way, so fix this first."
        )


def exchange(settings: Settings, redirect_url: str):
    from schwab import auth
    from schwab.auth import AuthContext, client_from_received_url

    state = parse_qs(urlparse(redirect_url).query).get("state", [None])[0]
    if not state:
        raise SystemExit("That URL has no ?state= — paste the full address you landed on")

    # Checked before the exchange, not after. An authorisation code lives about
    # thirty seconds and is single use, so discovering an unwritable token path
    # afterwards costs a whole round trip through the browser — which is exactly
    # what happened the first time, against a data/ directory owned by root
    # because Docker created it through the bind mount.
    ensure_token_path_writable(settings)
    # The library's own writer, so the token file keeps the shape its loader
    # expects rather than one guessed at here.
    write_token = getattr(auth, "__make_update_token_func")(str(settings.schwab_token_path))

    # Only the state is needed again; the authorisation URL is not, which is
    # what lets the two steps share nothing. Taking the state from the pasted
    # URL makes the CSRF check a formality — acceptable because the thing being
    # guarded against is a redirect you did not initiate, and you pasted this
    # one in by hand.
    context = AuthContext(settings.schwab_callback_url, None, state)
    return client_from_received_url(
        settings.schwab_api_key,
        settings.schwab_app_secret,
        context,
        redirect_url,
        write_token,
    )


def run_checks(client, settings: Settings) -> None:
    heading(f"REST quote for {settings.schwab_symbol}")
    quote = client.get_quote(settings.schwab_symbol)
    print(f"  HTTP {quote.status_code}")
    payload = quote.json() if quote.status_code == 200 else {}
    quotes_ok = False
    if payload:
        for symbol, data in list(payload.items())[:1]:
            values = data.get("quote", {})
            kind = data.get("assetMainType")
            description = (data.get("reference") or {}).get("description")
            print(f"  {symbol}: {kind} — {description}")
            print(f"  last={values.get('lastPrice')} bid={values.get('bidPrice')} "
                  f"ask={values.get('askPrice')}")
            # Schwab strips the leading slash and happily returns the equity of
            # the same name: /ES comes back as Eversource Energy at 72. A 200
            # with a body is not evidence of futures data, and treating it as
            # such is worse than a clean failure.
            quotes_ok = kind == "FUTURE"
        if not quotes_ok:
            print("  -> NOT futures. The slash was stripped and an equity")
            print("     returned in its place; REST futures quotes are unavailable.")
        else:
            print("  -> futures market data IS available over REST")
    else:
        print(f"  body: {quote.text[:200]}")
        print("  -> no quote returned")

    heading("Streamer bootstrap (/trader/v1/userPreference)")
    prefs = client.get_user_preferences()
    print(f"  HTTP {prefs.status_code}")
    streaming_ok = prefs.status_code == 200 and bool(prefs.json().get("streamerInfo"))
    if streaming_ok:
        print(f"  streamerInfo entries: {len(prefs.json()['streamerInfo'])}")
        print("  -> streaming is available; CHART_FUTURES should work")
    else:
        print(f"  body: {prefs.text[:200]}")
        print("  -> streaming is NOT available on this app")

    heading("Summary")
    print(f"  Quotes    : {'yes' if quotes_ok else 'no'}")
    print(f"  Streaming : {'yes' if streaming_ok else 'no'}")
    if quotes_ok and not streaming_ok:
        print("\n  Polling REST quotes is then the real-time path: it removes")
        print("  Yahoo's ten-minute delay without needing trading scope, at the")
        print("  cost of building bars from snapshots rather than receiving")
        print("  true exchange OHLCV.")


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--redirect-url", help="the full URL you were redirected to")
    args = parser.parse_args()
    settings = load_settings()

    try:
        from schwab.auth import client_from_token_file
    except ImportError:
        raise SystemExit("pip install -r requirements-dev.txt (schwab-py is not installed)")

    if args.redirect_url:
        heading("Exchanging the authorisation code")
        client = exchange(settings, args.redirect_url)
        print(f"  token written to {settings.schwab_token_path}")
    elif settings.schwab_token_path.exists():
        heading("Authentication")
        client = client_from_token_file(
            str(settings.schwab_token_path),
            settings.schwab_api_key,
            settings.schwab_app_secret,
        )
        print(f"  reused the token at {settings.schwab_token_path}")
    else:
        print_login_url(settings)
        sys.exit(0)

    run_checks(client, settings)


if __name__ == "__main__":
    main()