btc_weekly_report.py

#!/usr/bin/env python3
"""Weekly BTC DCA timing report.

Uses the existing 5am/5pm samples in state/btc-prices.jsonl.
Computes simple day-of-week stats for two candidate buy times:
- 05:00 local (hour==5)
- 17:00 local (hour==17)

Outputs:
- For each slot, which weekday had the lowest average BTC/USD and BTC/CAD
- Win-rate: how often each weekday was the lowest (within that week) for that slot

This is descriptive stats, not prediction.
"""

import json
import sys
from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime
from zoneinfo import ZoneInfo

TZ = ZoneInfo("America/Edmonton")
SLOTS = {
    5: "5am",
    17: "5pm",
}

@dataclass
class Rec:
    ts: int
    dt: datetime
    btc_usd: float
    btc_cad: float


def load_recs(path: str) -> list[Rec]:
    recs: list[Rec] = []
    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            obj = json.loads(line)
            ts = int(obj["ts"])
            dt = datetime.fromtimestamp(ts, TZ)
            recs.append(
                Rec(
                    ts=ts,
                    dt=dt,
                    btc_usd=float(obj["btc_usd"]),
                    btc_cad=float(obj["btc_cad"]),
                )
            )
    recs.sort(key=lambda r: r.ts)
    return recs


def week_key(dt: datetime) -> tuple[int, int]:
    # ISO week
    iso = dt.isocalendar()
    return (iso.year, iso.week)


def weekday_name(dt: datetime) -> str:
    return dt.strftime("%a")  # Mon, Tue...


def pick_latest_per_day_slot(recs: list[Rec], hour: int) -> dict[tuple[int, int], dict[str, Rec]]:
    """Return mapping week -> weekday -> record for that weekday/slot.

    If multiple samples exist in same weekday+slot, keep latest.
    """
    out: dict[tuple[int, int], dict[str, Rec]] = defaultdict(dict)
    for r in recs:
        if r.dt.hour != hour:
            continue
        wk = week_key(r.dt)
        wd = weekday_name(r.dt)
        prev = out[wk].get(wd)
        if prev is None or r.ts > prev.ts:
            out[wk][wd] = r
    return out


def summarize(slot_hour: int, by_week: dict[tuple[int, int], dict[str, Rec]]) -> str:
    slot_label = SLOTS.get(slot_hour, f"{slot_hour}:00")

    # Aggregate averages by weekday
    sums_usd = defaultdict(float)
    sums_cad = defaultdict(float)
    counts = defaultdict(int)

    # Win counts: within each week, which weekday had lowest price for that slot
    wins_usd = defaultdict(int)
    wins_cad = defaultdict(int)
    total_weeks = 0

    for wk, days in by_week.items():
        if len(days) < 4:
            # skip super sparse weeks
            continue
        total_weeks += 1

        # pick winner within this week for that slot
        usd_items = [(wd, rec.btc_usd) for wd, rec in days.items()]
        cad_items = [(wd, rec.btc_cad) for wd, rec in days.items()]
        usd_wd = min(usd_items, key=lambda x: x[1])[0]
        cad_wd = min(cad_items, key=lambda x: x[1])[0]
        wins_usd[usd_wd] += 1
        wins_cad[cad_wd] += 1

        for wd, rec in days.items():
            sums_usd[wd] += rec.btc_usd
            sums_cad[wd] += rec.btc_cad
            counts[wd] += 1

    if total_weeks == 0:
        return f"{slot_label}: not enough data yet."

    # Build sorted tables by average
    rows = []
    for wd in sorted(counts.keys()):
        n = counts[wd]
        avg_usd = sums_usd[wd] / n
        avg_cad = sums_cad[wd] / n
        winr_usd = wins_usd[wd] / total_weeks
        winr_cad = wins_cad[wd] / total_weeks
        rows.append((wd, n, avg_usd, avg_cad, winr_usd, winr_cad))

    rows_usd = sorted(rows, key=lambda r: r[2])
    rows_cad = sorted(rows, key=lambda r: r[3])

    best_usd = rows_usd[0]
    best_cad = rows_cad[0]

    lines = []
    lines.append(f"{slot_label} (weeks analyzed: {total_weeks})")
    lines.append(
        f"- Best avg BTC/USD: {best_usd[0]} (avg ${best_usd[2]:,.0f}, win-rate {best_usd[4]*100:.0f}%)"
    )
    lines.append(
        f"- Best avg BTC/CAD: {best_cad[0]} (avg C${best_cad[3]:,.0f}, win-rate {best_cad[5]*100:.0f}%)"
    )

    return "\n".join(lines)


def main():
    path = sys.argv[1] if len(sys.argv) > 1 else "/home/bot/Workspace/state/btc-prices.jsonl"
    recs = load_recs(path)

    if len(recs) < 10:
        print("Not much data yet (need a few weeks). I’ll start giving real weekday stats once we have more samples.")
        return

    parts = []
    parts.append("Weekly DCA timing stats (descriptive, not predictive)")

    for hour in (5, 17):
        by_week = pick_latest_per_day_slot(recs, hour)
        parts.append(summarize(hour, by_week))

    print("\n".join(parts))


if __name__ == "__main__":
    main()