Mercurial

comparison schwab_trader/schwab_cli.py @ 220:eb8b4230fdb9

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
[schwab-trader] Add guarded trading experiment
author MrJuneJune <me@mrjunejune.com>
date Sun, 02 Aug 2026 08:52:13 -0700
parents
children
comparison
equal deleted inserted replaced
214:4c725fde6999 220:eb8b4230fdb9
1 from __future__ import annotations
2
3 import argparse
4 import json
5 import sys
6
7 from schwab_trader.schwab_client import (
8 SchwabConfig,
9 SchwabError,
10 build_authorization_url,
11 build_equity_order,
12 exchange_code_for_tokens,
13 get_account,
14 get_account_numbers,
15 get_accounts,
16 load_tokens,
17 place_order,
18 refresh_tokens,
19 save_tokens,
20 )
21
22
23 def main(argv: list[str] | None = None) -> int:
24 parser = argparse.ArgumentParser(
25 description="Safe Schwab Trader API helper. This tool never chooses trades for you.",
26 )
27 subparsers = parser.add_subparsers(dest="command", required=True)
28
29 auth_url = subparsers.add_parser("auth-url", help="Print the Schwab OAuth authorization URL")
30 auth_url.add_argument("--state", help="Optional OAuth state value")
31
32 token = subparsers.add_parser("token", help="Exchange an authorization code/callback URL for tokens")
33 token.add_argument("--code", required=True, help="Authorization code or full callback URL")
34
35 subparsers.add_parser("refresh", help="Refresh and save tokens")
36
37 subparsers.add_parser("account-numbers", help="Print Schwab account-number/account-hash mapping")
38
39 accounts = subparsers.add_parser("accounts", help="Print account data")
40 accounts.add_argument("--positions", action="store_true", help="Include positions if API permissions allow it")
41
42 account = subparsers.add_parser("account", help="Print one account by account hash")
43 account.add_argument("--account-hash", required=True)
44 account.add_argument("--positions", action="store_true", help="Include positions if API permissions allow it")
45
46 build_order = subparsers.add_parser("build-equity-order", help="Build and print a stock order JSON payload")
47 _add_order_args(build_order)
48
49 place_equity = subparsers.add_parser(
50 "place-equity-order",
51 help="Place a user-specified stock order; defaults to dry-run output only",
52 )
53 _add_order_args(place_equity)
54 place_equity.add_argument("--account-hash", required=True)
55 place_equity.add_argument("--live", action="store_true", help="Actually submit the order to Schwab")
56 place_equity.add_argument(
57 "--confirm-live-trade",
58 action="store_true",
59 help="Required with --live to reduce accidental orders",
60 )
61
62 args = parser.parse_args(argv)
63
64 try:
65 if args.command == "auth-url":
66 config = SchwabConfig.from_env()
67 print(build_authorization_url(config.app_key, config.redirect_uri, args.state))
68 return 0
69
70 if args.command == "token":
71 config = SchwabConfig.from_env()
72 tokens = exchange_code_for_tokens(config, args.code)
73 save_tokens(config.token_file, tokens)
74 print(f"Saved tokens to {config.token_file}")
75 return 0
76
77 if args.command == "refresh":
78 config = SchwabConfig.from_env()
79 tokens = refresh_tokens(config)
80 save_tokens(config.token_file, tokens)
81 print(f"Refreshed tokens in {config.token_file}")
82 return 0
83
84 if args.command == "account-numbers":
85 config = SchwabConfig.from_env()
86 access_token = _load_access_token(config)
87 _print_json(get_account_numbers(access_token).body)
88 return 0
89
90 if args.command == "accounts":
91 config = SchwabConfig.from_env()
92 access_token = _load_access_token(config)
93 fields = "positions" if args.positions else None
94 _print_json(get_accounts(access_token, fields).body)
95 return 0
96
97 if args.command == "account":
98 config = SchwabConfig.from_env()
99 access_token = _load_access_token(config)
100 fields = "positions" if args.positions else None
101 _print_json(get_account(access_token, args.account_hash, fields).body)
102 return 0
103
104 if args.command == "build-equity-order":
105 order = _build_order_from_args(args)
106 _print_json(order)
107 return 0
108
109 if args.command == "place-equity-order":
110 order = _build_order_from_args(args)
111 if not args.live:
112 print("DRY RUN: order was not sent. Add --live --confirm-live-trade to submit.")
113 _print_json(order)
114 return 0
115 if not args.confirm_live_trade:
116 raise SchwabError("--live requires --confirm-live-trade")
117
118 config = SchwabConfig.from_env()
119 access_token = _load_access_token(config)
120 response = place_order(access_token, args.account_hash, order)
121 print(f"Schwab order response status: {response.status}")
122 location = response.headers.get("Location")
123 if location:
124 print(f"Order location: {location}")
125 if response.body is not None:
126 _print_json(response.body)
127 return 0
128
129 raise SchwabError(f"Unknown command: {args.command}")
130 except SchwabError as error:
131 print(f"error: {error}", file=sys.stderr)
132 return 1
133
134
135 def _add_order_args(parser: argparse.ArgumentParser) -> None:
136 parser.add_argument("--action", required=True, choices=["BUY", "SELL", "buy", "sell"])
137 parser.add_argument("--symbol", required=True)
138 parser.add_argument("--quantity", required=True, type=float)
139 parser.add_argument("--order-type", default="MARKET", choices=["MARKET", "LIMIT", "market", "limit"])
140 parser.add_argument("--price", type=float, help="Required for LIMIT orders; invalid for MARKET orders")
141 parser.add_argument("--duration", default="DAY")
142 parser.add_argument("--session", default="NORMAL")
143
144
145 def _build_order_from_args(args: argparse.Namespace) -> dict:
146 return build_equity_order(
147 action=args.action,
148 symbol=args.symbol,
149 quantity=args.quantity,
150 order_type=args.order_type,
151 price=args.price,
152 duration=args.duration,
153 session=args.session,
154 )
155
156
157 def _load_access_token(config: SchwabConfig) -> str:
158 tokens = load_tokens(config.token_file)
159 access_token = tokens.get("access_token")
160 if not access_token:
161 raise SchwabError(f"No access_token in {config.token_file}")
162 return access_token
163
164
165 def _print_json(value: object) -> None:
166 print(json.dumps(value, indent=2, sort_keys=True))
167
168
169 if __name__ == "__main__":
170 raise SystemExit(main())
171