diff --git a/scripts/check_schwab.py b/scripts/check_schwab.py index eef71d9..db446b6 100644 --- a/scripts/check_schwab.py +++ b/scripts/check_schwab.py @@ -1,23 +1,33 @@ """Find out what a Schwab app is actually entitled to, before building on it. -Run once after putting SCHWAB_API_KEY and SCHWAB_APP_SECRET in .env. It answers -the three questions that decide the design, and it answers them empirically -rather than from documentation: +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 here. + for Market Data Production alone is expected to fail there. -Uses the manual OAuth flow because the callback is hosted rather than local: -visit the printed URL, complete the login, then paste the URL you land on. The -callback page shows it ready to copy. +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 asyncio +import argparse +import sys +from urllib.parse import parse_qs, urlparse from app.config import Settings @@ -26,71 +36,187 @@ def heading(text: str) -> None: print(f"\n{text}\n{'-' * len(text)}") -async def main() -> None: +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: - from schwab.auth import client_from_manual_flow, client_from_token_file - except ImportError: - raise SystemExit("pip install -r requirements-dev.txt (schwab-py is not installed)") - - heading("1. Authentication") - token_path = str(settings.schwab_token_path) - try: - client = client_from_token_file( - token_path, settings.schwab_api_key, settings.schwab_app_secret - ) - print(f" reused the token at {token_path}") + response = httpx.get(authorization_url, follow_redirects=False, timeout=15) except Exception: - print(" no usable token — starting the manual flow") - client = client_from_manual_flow( - settings.schwab_api_key, - settings.schwab_app_secret, - settings.schwab_callback_url, - token_path, - ) - print(" authenticated") + return True # Network trouble is not evidence about the key. + return "invalid_client" not in response.text - heading(f"2. REST quote for {settings.schwab_symbol}") - response = client.get_quote(settings.schwab_symbol) - print(f" HTTP {response.status_code}") - if response.status_code == 200: - payload = response.json() - print(f" keys: {list(payload)[:4]}") - for symbol, data in list(payload.items())[:1]: - quote = data.get("quote", {}) - print(f" {symbol}: last={quote.get('lastPrice')} " - f"bid={quote.get('bidPrice')} ask={quote.get('askPrice')}") + +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 ''") + + +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") + + settings.schwab_token_path.parent.mkdir(parents=True, exist_ok=True) + # 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}") + quotes_ok = quote.status_code == 200 and bool(quote.json()) + if quotes_ok: + for symbol, data in list(quote.json().items())[:1]: + values = data.get("quote", {}) + print(f" {symbol}: last={values.get('lastPrice')} " + f"bid={values.get('bidPrice')} ask={values.get('askPrice')}") print(" -> futures market data IS available") else: - print(f" body: {response.text[:200]}") + print(f" body: {quote.text[:200]}") print(" -> futures market data is NOT available on this app") - heading("3. Streamer bootstrap (/trader/v1/userPreference)") + heading("Streamer bootstrap (/trader/v1/userPreference)") prefs = client.get_user_preferences() print(f" HTTP {prefs.status_code}") - if prefs.status_code == 200: - info = prefs.json().get("streamerInfo") or [] - print(f" streamerInfo entries: {len(info)}") + 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. This endpoint belongs to the") - print(" Accounts and Trading product; a Market Data Production app") - print(" cannot reach it, so CHART_FUTURES is out of reach until the") - print(" app adds that product.") + print(" -> streaming is NOT available on this app") heading("Summary") - print(" Quotes available :", response.status_code == 200) - print(" Streaming available:", prefs.status_code == 200) - if response.status_code == 200 and prefs.status_code != 200: + 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__": - asyncio.run(main()) + main()