125 lines
3.3 KiB
Python
125 lines
3.3 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
market_watchlist.py
|
||
|
|
|
||
|
|
Maintain a local watchlist and summarize current quotes.
|
||
|
|
|
||
|
|
Usage:
|
||
|
|
python scripts/market_watchlist.py add AAPL MSFT USD/ZAR
|
||
|
|
python scripts/market_watchlist.py remove MSFT
|
||
|
|
python scripts/market_watchlist.py list
|
||
|
|
python scripts/market_watchlist.py summary
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
from typing import List
|
||
|
|
|
||
|
|
WATCHLIST_PATH = os.path.join(".cache", "market-tracker", "watchlist.json")
|
||
|
|
os.makedirs(os.path.dirname(WATCHLIST_PATH), exist_ok=True)
|
||
|
|
|
||
|
|
|
||
|
|
def load_watchlist() -> List[str]:
|
||
|
|
if not os.path.exists(WATCHLIST_PATH):
|
||
|
|
return []
|
||
|
|
with open(WATCHLIST_PATH, "r", encoding="utf-8") as f:
|
||
|
|
data = json.load(f)
|
||
|
|
items = data.get("items") or []
|
||
|
|
# de-dupe, preserve order
|
||
|
|
seen = set()
|
||
|
|
out = []
|
||
|
|
for x in items:
|
||
|
|
x = str(x).strip()
|
||
|
|
if x and x not in seen:
|
||
|
|
out.append(x)
|
||
|
|
seen.add(x)
|
||
|
|
return out
|
||
|
|
|
||
|
|
|
||
|
|
def save_watchlist(items: List[str]) -> None:
|
||
|
|
with open(WATCHLIST_PATH, "w", encoding="utf-8") as f:
|
||
|
|
json.dump({"items": items}, f, ensure_ascii=False, indent=2)
|
||
|
|
|
||
|
|
|
||
|
|
def cmd_add(symbols: List[str]) -> None:
|
||
|
|
items = load_watchlist()
|
||
|
|
for s in symbols:
|
||
|
|
s = s.strip()
|
||
|
|
if s and s not in items:
|
||
|
|
items.append(s)
|
||
|
|
save_watchlist(items)
|
||
|
|
print(json.dumps({"ok": True, "items": items}, indent=2))
|
||
|
|
|
||
|
|
|
||
|
|
def cmd_remove(symbols: List[str]) -> None:
|
||
|
|
items = load_watchlist()
|
||
|
|
rm = set(s.strip() for s in symbols if s.strip())
|
||
|
|
items = [x for x in items if x not in rm]
|
||
|
|
save_watchlist(items)
|
||
|
|
print(json.dumps({"ok": True, "items": items}, indent=2))
|
||
|
|
|
||
|
|
|
||
|
|
def cmd_list() -> None:
|
||
|
|
print(json.dumps({"items": load_watchlist()}, indent=2))
|
||
|
|
|
||
|
|
|
||
|
|
def cmd_summary() -> None:
|
||
|
|
items = load_watchlist()
|
||
|
|
if not items:
|
||
|
|
print(json.dumps({"items": [], "summary": []}, indent=2))
|
||
|
|
return
|
||
|
|
|
||
|
|
results = []
|
||
|
|
for sym in items:
|
||
|
|
# Call market_quote.py to keep logic centralized
|
||
|
|
p = subprocess.run(
|
||
|
|
[sys.executable, os.path.join(os.path.dirname(__file__), "market_quote.py"), sym],
|
||
|
|
capture_output=True,
|
||
|
|
text=True,
|
||
|
|
)
|
||
|
|
if p.returncode != 0:
|
||
|
|
results.append({"symbol": sym, "error": p.stderr.strip() or p.stdout.strip()})
|
||
|
|
continue
|
||
|
|
try:
|
||
|
|
results.append(json.loads(p.stdout))
|
||
|
|
except Exception:
|
||
|
|
results.append({"symbol": sym, "error": "Invalid JSON from market_quote.py", "raw": p.stdout[:500]})
|
||
|
|
|
||
|
|
print(json.dumps({"items": items, "summary": results}, ensure_ascii=False, indent=2))
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> None:
|
||
|
|
ap = argparse.ArgumentParser()
|
||
|
|
sub = ap.add_subparsers(dest="cmd", required=True)
|
||
|
|
|
||
|
|
sp_add = sub.add_parser("add")
|
||
|
|
sp_add.add_argument("symbols", nargs="+")
|
||
|
|
|
||
|
|
sp_rm = sub.add_parser("remove")
|
||
|
|
sp_rm.add_argument("symbols", nargs="+")
|
||
|
|
|
||
|
|
sub.add_parser("list")
|
||
|
|
sub.add_parser("summary")
|
||
|
|
|
||
|
|
args = ap.parse_args()
|
||
|
|
|
||
|
|
if args.cmd == "add":
|
||
|
|
cmd_add(args.symbols)
|
||
|
|
elif args.cmd == "remove":
|
||
|
|
cmd_remove(args.symbols)
|
||
|
|
elif args.cmd == "list":
|
||
|
|
cmd_list()
|
||
|
|
elif args.cmd == "summary":
|
||
|
|
cmd_summary()
|
||
|
|
else:
|
||
|
|
raise SystemExit("Unknown command")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|