#!/usr/bin/env python3 """Weekly prep — Friday close snapshot. Writes 03-research/weekly/-week-of.md with: - Week-of date in frontmatter - Full universe weekly performance table (1d, 1w, 1m) - This week's calendar events (Tier 1/2/structural) for the week - COT positioning summary (latest available) - Regime dashboard pre-filled with direction arrows based on price action - Empty placeholders for the user's regime read, what-changed, book update The user then edits the file in Obsidian to add their own commentary. """ from __future__ import annotations import argparse import sys from datetime import date, datetime, timedelta from pathlib import Path from zoneinfo import ZoneInfo VAULT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(VAULT / "scripts")) from data.prices import all_change_metrics # noqa: E402 from data.calendar_parse import parse_events # noqa: E402 from data.cot import fetch_universe_cot, cot_summary # noqa: E402 from data.universe import BUCKET_LABELS # noqa: E402 WEEKLY_DIR = VAULT / "03-research" / "weekly" def fmt_change(pct: float | None) -> str: if pct is None: return "—" sign = "+" if pct >= 0 else "" return f"{sign}{pct:.2f}%" def fmt_last(price: float | None) -> str: if price is None: return "—" if abs(price) >= 1000: return f"{price:,.0f}" return f"{price:,.2f}" def arrow(pct: float | None) -> str: if pct is None or pct == 0: return "→" return "↑" if pct > 0 else "↓" def render_universe_table(metrics: list[dict]) -> str: lines = [ "| Ticker | Bucket | Name | Last | 1w % | 1m % |", "|---|---|---|---:|---:|---:|", ] # Sort by bucket, then by name metrics_sorted = sorted(metrics, key=lambda m: (m.get("bucket", ""), m.get("name", ""))) for m in metrics_sorted: bucket_label = BUCKET_LABELS.get(m.get("bucket", ""), m.get("bucket", "")) lines.append( f"| {m['display']} | {bucket_label} | {m.get('name', '')} | " f"{fmt_last(m['last'])} | {fmt_change(m['chg_5d_pct'])} | " f"{fmt_change(m['chg_1m_pct'])} |" ) return "\n".join(lines) def render_week_calendar(week_events: list[dict], week_start: date, week_end: date) -> str: if not week_events: return f"_No events in week {week_start} – {week_end}._" tz = ZoneInfo("Europe/Zurich") lines = [ "| Date | Time (Zurich) | Tier | Event |", "|---|---|---|---|", ] for e in week_events: zurich = e["dtstart_utc"].astimezone(tz) d = zurich.strftime("%a %m-%d") t = zurich.strftime("%H:%M") tier = f"T{e['tier']}" if isinstance(e["tier"], int) else (e["tier"] or "—") sm = e["summary"] for tag in ("[T1]", "[T2]", "[STRUCTURAL]"): sm = sm.replace(tag, "").strip() lines.append(f"| {d} | {t} | {tier} | {sm} |") return "\n".join(lines) def render_cot_table(cot_rows: dict) -> str: lines = [ "| Instrument | CFTC date | Noncomm net | Open interest |", "|---|---|---:|---:|", ] for iid, row in cot_rows.items(): if row is None: lines.append(f"| {iid} | no data | — | — |") continue rd = row.get("report_date_as_yyyy_mm_dd", "?")[:10] try: ncl = int(row.get("noncomm_positions_long_all", 0)) ncs = int(row.get("noncomm_positions_short_all", 0)) ncnet = ncl - ncs ncnet_s = f"{ncnet:+,}" except (TypeError, ValueError): ncnet_s = "?" try: oi = int(row.get("open_interest_all", 0)) oi_s = f"{oi:,}" except (TypeError, ValueError): oi_s = "?" lines.append(f"| {iid} | {rd} | {ncnet_s} | {oi_s} |") return "\n".join(lines) def regime_arrow(metrics: list[dict], ticker: str, period_key: str) -> str: """Return arrow for regime dashboard based on price change of a representative ticker.""" for m in metrics: if m["display"] == ticker or m["ticker"] == ticker: pct = m.get(period_key) return arrow(pct) return "→" def render_regime_dashboard(metrics: list[dict]) -> str: """Pre-fill regime dashboard with arrows based on 1m price action of representative tickers.""" growth = regime_arrow(metrics, "SPY", "chg_1m_pct") # S&P 500 inflation = regime_arrow(metrics, "GLD", "chg_1m_pct") # Gold (proxy for inflation hedge demand) policy = regime_arrow(metrics, "TLT", "chg_1m_pct") # Long bonds (price up = yields down = dovish lean) positioning = regime_arrow(metrics, "^VIX", "chg_1m_pct") # VIX down = risk-on return ( "| Axis | Direction | One-line read |\n" "|---|---|---|\n" f"| Growth | {growth} | _your read_ |\n" f"| Inflation | {inflation} | _your read_ |\n" f"| Policy | {policy} | _your read_ |\n" f"| Positioning | {positioning} | _your read_ |\n" ) def render_weekly(target_friday: date, metrics: list[dict], week_events: list[dict], cot_rows: dict) -> str: week_start = target_friday - timedelta(days=4) # Mon week_end = target_friday # Fri date_str = target_friday.isoformat() frontmatter = ( "---\n" f"type: weekly-note\n" f"date: {date_str}\n" f"week-of: {week_start.isoformat()}\n" f"week-end: {week_end.isoformat()}\n" f"status: prepopulated\n" f"generated: {datetime.now().isoformat(timespec='seconds')}\n" f"location: Zurich, Switzerland (Europe/Zurich)\n" "---\n" ) title = f"# Weekly Note — Week of {week_start} (closes {date_str})\n" parts = [ frontmatter, title, "## 1. Regime dashboard (pre-filled, edit)\n", render_regime_dashboard(metrics), "", "## 2. What changed this week (your read)\n", "_3-4 paragraphs. Data vs market pricing. What's the gap, and is it closing or widening?_\n", "## 3. COT / positioning corner\n", render_cot_table(cot_rows), "", "_CTA positioning estimate: derive from COT extremes + your read._\n", "## 4. Chart of the week\n", "_Embed chart from `_assets/`. Apply the [[../../00-methodology/decision-process]] checklist._\n", "## 5. Book update\n", "- Positions: _list with current P&L_", "- P&L MTD: _bp of NAV_", "- Changes this week: _add/trim/exit with reason (link to trade folders)_", "- Realised vs unrealised split: _\n", "## 6. Week ahead\n", "### This week's events\n", render_week_calendar(week_events, week_start, week_end), "", "### Full universe weekly performance\n", render_universe_table(metrics), "", "## OpenTimestamps hash (when publication track opens)\n", "`pending`\n", "## Pseudonym footer\n", "`pending`\n", ] return "\n".join(parts) def main(): parser = argparse.ArgumentParser() parser.add_argument( "--date", default=None, help="Target Friday YYYY-MM-DD (default: most recent Friday in Europe/Zurich)" ) parser.add_argument( "--force", action="store_true", help="Overwrite existing weekly note" ) parser.add_argument( "--skip-cot", action="store_true", help="Skip COT fetch (faster, used for tests)" ) args = parser.parse_args() if args.date: target = date.fromisoformat(args.date) else: # Most recent Friday (or today if Friday) today = datetime.now(ZoneInfo("Europe/Zurich")).date() days_since_friday = (today.weekday() - 4) % 7 target = today - timedelta(days=days_since_friday) out_path = WEEKLY_DIR / f"{target.isoformat()}-week-of.md" if out_path.exists() and not args.force: print(f"EXISTS: {out_path} (use --force to overwrite)") return 0 print(f"Fetching full universe metrics for {target}...") metrics = all_change_metrics() print(f" {len(metrics)} tickers") week_start_dt = datetime(target.year, target.month, target.day, tzinfo=ZoneInfo("Europe/Zurich")) - timedelta(days=4) week_end_dt = week_start_dt + timedelta(days=5) print(f"Fetching calendar events for week {week_start_dt.date()} – {week_end_dt.date()}...") week_events = parse_events(window_start=week_start_dt.astimezone(__import__('datetime').timezone.utc), window_end=week_end_dt.astimezone(__import__('datetime').timezone.utc)) print(f" {len(week_events)} events") if args.skip_cot: cot_rows = {} else: print("Fetching COT positioning...") cot_rows = fetch_universe_cot() print(f" {len(cot_rows)} instruments") body = render_weekly(target, metrics, week_events, cot_rows) out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_text(body) print(f"Wrote: {out_path} ({len(body)} bytes)") return 0 if __name__ == "__main__": sys.exit(main())