如何用回测验证YouTube上的投资论断 — 从字幕到免费行情API

# 安装（二选一）
brew install yt-dlp        # macOS
pip install yt-dlp         # Python 环境

# 跳过视频，只取韩语和英语字幕
yt-dlp --skip-download \
       --write-auto-sub --write-sub \
       --sub-lang "ko,en" --sub-format vtt \
       -o "vid.%(ext)s" "<视频地址>"

# 结果: vid.ko.vtt

import re

lines, prev = [], None
for line in open("vid.ko.vtt", encoding="utf-8"):
    # 丢弃时间戳行和头部
    if "-->" in line or line.startswith(("WEBVTT", "Kind:", "Language:")):
        continue
    text = re.sub(r"<[^>]+>", "", line).strip()   # 去掉标签
    if text and text != prev:                      # 去掉连续重复
        lines.append(text)
        prev = text

open("transcript.txt", "w", encoding="utf-8").write("\n".join(lines))
print(len(lines), "行")

import json, urllib.request

UA = {"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
                    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126 Safari/537.36"}

def daily(symbol, asset="stocks", frm="2016-01-01", to="2026-07-31"):
    url = (f"https://api.nasdaq.com/api/quote/{symbol}/historical"
           f"?assetclass={asset}&fromdate={frm}&todate={to}&limit=99999")
    req = urllib.request.Request(url, headers=UA)
    rows = json.load(urllib.request.urlopen(req, timeout=30))
    rows = rows["data"]["tradesTable"]["rows"]

    out = []
    for r in rows:
        close = float(r["close"].replace("$", "").replace(",", ""))   # "$207.12" → 207.12
        mm, dd, yy = r["date"].split("/")                              # "07/30/2026"
        out.append({"d": f"{yy}-{mm}-{dd}", "c": close})
    out.sort(key=lambda x: x["d"])
    return out

bars = daily("MU")
print(len(bars), bars[-1])

import re, urllib.request

UA = {"User-Agent": "Mozilla/5.0"}

def daily_kr(code, start="20160101", end="20260731"):
    url = ("https://api.finance.naver.com/siseJson.naver"
           f"?symbol={code}&requestType=1&startTime={start}&endTime={end}&timeframe=day")
    req = urllib.request.Request(url, headers=UA)
    text = urllib.request.urlopen(req, timeout=30).read().decode("utf-8")

    # 不是标准JSON而是单引号数组，用正则提取
    # ['20260730', 开盘, 最高, 最低, 收盘, 成交量, 外资持股率]
    rows = re.findall(r"\['(\d{8})',\s*[\d.]+,\s*[\d.]+,\s*[\d.]+,\s*([\d.]+)", text)
    out = [{"d": f"{d[:4]}-{d[4:6]}-{d[6:]}", "c": float(c)} for d, c in rows]
    out.sort(key=lambda x: x["d"])
    return out

print(daily_kr("005930")[-1])   # 三星电子

import statistics as st

def backtest(bars, dd_thresh=-0.35, pop=0.20, horizon=63):
    """从60日高点跌破dd_thresh后，单日上涨pop以上 → 当日收盘买入"""
    signals, baseline = [], []

    for i in range(60, len(bars) - horizon):
        c, prev = bars[i]["c"], bars[i - 1]["c"]
        high60 = max(b["c"] for b in bars[i - 60:i + 1])
        fwd = bars[i + horizon]["c"] / c - 1        # horizon日后的收益率

        baseline.append(fwd * 100)                   # 对照组：所有交易日
        if c / prev - 1 >= pop and c / high60 - 1 <= dd_thresh:
            signals.append(fwd * 100)                # 只取信号出现日

    return signals, baseline

sig, base = backtest(bars)
print(f"信号   n={len(sig):4d}  平均 {st.mean(sig):6.1f}%  中位数 {st.median(sig):6.1f}%")
print(f"对照组 n={len(base):4d}  平均 {st.mean(base):6.1f}%  中位数 {st.median(base):6.1f}%")
# 信号的平均值若超不过对照组，这条规则就没有实力