Chris Amow
5ccb2bfb52
config.py had no Schwab fields at all, so the keys listed in .env.example were being silently dropped by extra="ignore". They exist now, blank, and nothing reads them while live_source is yahoo. The token path moves under data/, which is the Coolify persistent volume. Left at the repository root it would vanish on every rebuild, and re-authenticating is an interactive browser flow, not something a deploy can do for itself. scripts/check_schwab.py answers empirically what the app is entitled to rather than inferring it from documentation: whether the credentials authenticate, whether /ES quotes return (futures market data is a separate entitlement from equities), and whether the streamer bootstrap responds. That last one is the decision. StreamClient.login() reads /trader/v1/userPreference for its socket URL and credentials, and that path belongs to the Accounts and Trading product — so an app registered for Market Data Production alone cannot stream, and CHART_FUTURES is unreachable until the app adds it. The script reports which of the two paths is open instead of leaving it to be discovered halfway through an implementation. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
96 lines
3.9 KiB
Python
96 lines
3.9 KiB
Python
"""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:
|
|
|
|
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.
|
|
|
|
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.
|
|
|
|
python3 -m scripts.check_schwab
|
|
"""
|
|
import asyncio
|
|
|
|
from app.config import Settings
|
|
|
|
|
|
def heading(text: str) -> None:
|
|
print(f"\n{text}\n{'-' * len(text)}")
|
|
|
|
|
|
async def main() -> None:
|
|
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")
|
|
|
|
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}")
|
|
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")
|
|
|
|
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')}")
|
|
print(" -> futures market data IS available")
|
|
else:
|
|
print(f" body: {response.text[:200]}")
|
|
print(" -> futures market data is NOT available on this app")
|
|
|
|
heading("3. 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)}")
|
|
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.")
|
|
|
|
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("\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.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|