sim_random10.py

"""
One-shot simulation: pick 10 random TW stocks, fetch predictor signal,
then compute P&L for two strategies over the past ~1 month:

  A) Equal weight across ALL 10 (random baseline)
  B) Equal weight ONLY across stocks currently flagged BUY by the predictor

Uses the live Render API so this reflects the same model the product serves.
"""

import random
import time
import json
import sys
from datetime import date, timedelta

import requests

if sys.stdout.encoding and sys.stdout.encoding.lower() != "utf-8":
    try:
        sys.stdout.reconfigure(encoding="utf-8")
    except Exception:
        pass

BASE = "https://stock-predictor-1oyc.onrender.com"

UNIVERSE = [
    ("2330", "台積電"), ("2317", "鴻海"), ("2454", "聯發科"),
    ("2881", "富邦金"), ("2882", "國泰金"), ("2412", "中華電"),
    ("1301", "台塑"),   ("1303", "南亞"),   ("2002", "中鋼"),
    ("2303", "聯電"),   ("2308", "台達電"), ("2891", "中信金"),
    ("3008", "大立光"), ("2603", "長榮"),   ("2609", "陽明"),
    ("2615", "萬海"),   ("3034", "聯詠"),   ("6505", "台塑化"),
    ("1216", "統一"),   ("2885", "元大金"), ("2884", "玉山金"),
    ("2886", "兆豐金"), ("2892", "第一金"), ("2357", "華碩"),
    ("2382", "廣達"),   ("6669", "緯穎"),   ("3711", "日月光投控"),
    ("2912", "統一超"), ("1101", "台泥"),   ("2880", "華南金"),
    ("3231", "緯創"),   ("1476", "儒鴻"),   ("2207", "和泰車"),
    ("4904", "遠傳"),   ("3045", "台灣大"), ("2379", "瑞昱"),
]


def pick_random(n: int = 10, seed: int | None = None) -> list[tuple[str, str]]:
    rng = random.Random(seed)
    return rng.sample(UNIVERSE, n)


def fetch_history(code: str, months: int = 2) -> list[dict]:
    r = requests.get(f"{BASE}/api/stock/{code}/history", params={"months": months}, timeout=60)
    r.raise_for_status()
    return r.json().get("data", [])


def fetch_batch_predict(codes: list[str], retries: int = 3) -> list[dict]:
    last_exc = None
    for attempt in range(retries):
        try:
            r = requests.get(f"{BASE}/api/stock/batch", params={"codes": ",".join(codes)}, timeout=180)
            r.raise_for_status()
            return r.json().get("results", [])
        except Exception as exc:
            last_exc = exc
            print(f"    retry {attempt + 1}/{retries}: {exc}")
            time.sleep(5)
    raise last_exc


def get_prices_around(hist: list[dict], target_days_ago: int = 21) -> tuple[float, float, str, str]:
    """Return (buy_price, sell_price, buy_date, sell_date) using close prices."""
    if not hist:
        raise ValueError("no history")
    sell = hist[-1]
    target_idx = max(0, len(hist) - 1 - target_days_ago)
    buy = hist[target_idx]
    return float(buy["close"]), float(sell["close"]), buy["date"], sell["date"]


def main():
    seed_arg = sys.argv[1] if len(sys.argv) > 1 else None
    seed = int(seed_arg) if seed_arg else int(time.time()) % 10000
    picked = pick_random(10, seed=seed)
    print(f"=== seed={seed} — randomly picked 10 TW stocks ===")
    for code, name in picked:
        print(f"  {code} {name}")
    print()

    # Fetch history for each
    rows = []
    for code, name in picked:
        try:
            hist = fetch_history(code, months=2)
            buy_p, sell_p, buy_d, sell_d = get_prices_around(hist, target_days_ago=21)
            pct = (sell_p - buy_p) / buy_p * 100
            rows.append({
                "code": code, "name": name,
                "buy_date": buy_d, "sell_date": sell_d,
                "buy_price": buy_p, "sell_price": sell_p,
                "pct": pct,
            })
            print(f"  history {code} {name}: {buy_d} {buy_p:.2f} -> {sell_d} {sell_p:.2f}  ({pct:+.2f}%)")
        except Exception as exc:
            print(f"  history {code} FAILED: {exc}")
        time.sleep(0.3)
    print()

    # Fetch predictor signals (batch, 5 at a time, rate limited)
    preds: dict[str, dict] = {}
    codes = [r["code"] for r in rows]
    for i in range(0, len(codes), 5):
        chunk = codes[i : i + 5]
        try:
            res = fetch_batch_predict(chunk)
            for p in res:
                if "error" not in p:
                    preds[p["code"]] = p
            print(f"  predict batch {i//5 + 1}: {[(p['code'], p.get('signal'), p.get('buy_prob')) for p in res]}")
        except Exception as exc:
            print(f"  predict batch FAILED: {exc}")
        if i + 5 < len(codes):
            time.sleep(15)  # batch rate limit: 5/min
    print()

    # Simulate
    capital = 1_000_000

    # Strategy A: equal weight, all 10
    if rows:
        per = capital / len(rows)
        total_a = sum(per * (1 + r["pct"] / 100) for r in rows)
        pnl_a = total_a - capital
    else:
        total_a = capital
        pnl_a = 0

    # Strategy B: equal weight, only BUY signals
    buys = [r for r in rows if preds.get(r["code"], {}).get("signal") == "BUY"]
    if buys:
        per_b = capital / len(buys)
        total_b = sum(per_b * (1 + r["pct"] / 100) for r in buys)
        pnl_b = total_b - capital
    else:
        total_b = capital
        pnl_b = 0

    print("=" * 60)
    print(f"本金: NT${capital:,}")
    print()
    print(f"[A] 隨機等權重 10 檔")
    for r in rows:
        per = capital / len(rows)
        print(f"  {r['code']} {r['name']:8} {r['pct']:+6.2f}%  損益 {per * r['pct'] / 100:+10,.0f}")
    print(f"  -> 總市值 NT${total_a:,.0f}  損益 NT${pnl_a:+,.0f}  ({pnl_a/capital*100:+.2f}%)")
    print()
    print(f"[B] 只買預測 BUY 的 {len(buys)} 檔 (等權重)")
    if buys:
        for r in buys:
            per_b = capital / len(buys)
            print(f"  {r['code']} {r['name']:8} {r['pct']:+6.2f}%  損益 {per_b * r['pct'] / 100:+10,.0f}")
        print(f"  -> 總市值 NT${total_b:,.0f}  損益 NT${pnl_b:+,.0f}  ({pnl_b/capital*100:+.2f}%)")
    else:
        print("  (預測器目前對這 10 檔都沒有 BUY 訊號 -> 全部持幣)")
        print(f"  -> 總市值 NT${total_b:,.0f}  損益 NT${pnl_b:+,.0f}  (0.00%)")

    print()
    print("註:B 策略使用「當下」預測器訊號回看一個月,有 look-ahead bias。")
    print("正統的做法是用 /api/stock/{code}/backtest 做 walk-forward。")


if __name__ == "__main__":
    main()