第 7 篇：多周期：周线在涨，小时线在跌，听谁的

default = day[cols + ["volume"]].resample("W").agg(B.OHLCV)
weekly = B.resample_ohlcv(day, "W-MON")
print("pandas 默认 \"W\"：", [t.strftime("%Y-%m-%d %a") for t in default.index[:3]])
print("\"W-MON\" + 左闭左标：", [t.strftime("%Y-%m-%d %a") for t in weekly.index[:3]])
print("Binance 官方周线：", [t.strftime("%Y-%m-%d %a") for t in official_1w.index[:3]])
print(pd.concat([official_1w[cols].head(2), default[cols].head(2)], keys=["官方", "默认 W"]))

pandas 默认 "W"： ['2017-08-20 Sun', '2017-08-27 Sun', '2017-09-03 Sun']
"W-MON" + 左闭左标： ['2017-08-14 Mon', '2017-08-21 Mon', '2017-08-28 Mon']
Binance 官方周线： ['2017-08-14 Mon', '2017-08-21 Mon', '2017-08-28 Mon']
                                   open     high     low    close
     time                                                        
官方   2017-08-14 00:00:00+00:00  4261.48  4485.39  3850.0  4086.29
     2017-08-21 00:00:00+00:00  4069.13  4453.91  3400.0  4310.01
默认 W 2017-08-20 00:00:00+00:00  4261.48  4485.39  3850.0  4086.29
     2017-08-27 00:00:00+00:00  4069.13  4453.91  3400.0  4310.01

W-MON 合成的周线和官方周线：共同的 456 周里对不上 0 周

floor("7D") 的结果： ['2017-08-17 Thu', '2017-08-17 Thu', '2017-08-24 Thu', '2017-08-24 Thu']

rebuilt = B.resample_ohlcv(minute, "4h", traded_only=True)
both = official_4h[cols].join(rebuilt[cols], rsuffix="_合成", how="outer")
wrong = pd.DataFrame({c: (both[c] - both[c + "_合成"]).abs() > 1e-8 for c in cols}).any(axis=1)
print("官方", len(official_4h), "根，合成", len(rebuilt), "根，对不上", wrong.sum(), "根：", wrong.groupby(both.index.year).sum()[lambda s: s > 0].to_dict())
offset = ((both.index >= "2017-12-04") & (both.index < "2017-12-19")) | ((both.index >= "2018-02-09") & (both.index < "2018-02-11"))
print("其中落在两段时间戳偏移时期的", (wrong & offset).sum(), "根；其余", [t.strftime("%Y-%m-%d %H:%M") for t in both.index[wrong & ~offset]])

官方 19794 根，合成 19794 根，对不上 89 根： {2017: 81, 2018: 6, 2020: 1, 2021: 1}
其中落在两段时间戳偏移时期的 87 根；其余 ['2020-12-21 12:00', '2021-04-22 00:00']

days_per_week = spy["close"].groupby(pd.Grouper(freq="W-MON", label="left", closed="left")).count()
print(days_per_week.value_counts().sort_index().to_dict())
print(days_per_week[days_per_week < 5].tail(3))

{2: 2, 4: 97, 5: 424}
date
2026-06-29    4
2026-09-07    4
2026-09-14    2
Name: close, dtype: int64

for name, df, periods in [("4 小时", four, 6 * 365), ("日线", day, 365), ("周线", weekly, 52)]:
    r = df["close"].pct_change().dropna()
    print(f"{name}：收益率标准差 {r.std():.2%}，年化 {r.std() * math.sqrt(periods):.1%}")

4 小时：收益率标准差 1.46%，年化 68.2%
日线：收益率标准差 3.51%，年化 67.0%
周线：收益率标准差 9.35%，年化 67.4%

weekly["green"] = weekly["close"] > weekly["open"]
week_of = pd.Grouper(freq="W-MON", label="left", closed="left")
red_days = (day["close"] < day["open"]).groupby(week_of).sum()
n_days = day["close"].groupby(week_of).count()
full = weekly[n_days == 7]
g = full[full["green"]]
print(f"完整的周阳线 {len(g)} 根：平均含阴线日 {red_days[g.index].mean():.2f} 天；"
      f"至少 3 天阴线 {(red_days[g.index] >= 3).mean():.1%}；一天阴线都没有 {(red_days[g.index] == 0).mean():.1%}")
red_4h = (four["close"] < four["open"]).groupby(week_of).mean()
print(f"周阳线里 4 小时阴线的比例 {red_4h[g.index].mean():.1%}；周阴线里 {red_4h[full[~full['green']].index].mean():.1%}")
dip = g["low"] / g["open"] - 1
print(f"周阳线从开盘向下最深跌了多少：中位数 {dip.median():.2%}，超过 3% 的 {(dip < -0.03).mean():.1%}")

完整的周阳线 250 根：平均含阴线日 2.76 天；至少 3 天阴线 62.8%；一天阴线都没有 1.2%
周阳线里 4 小时阴线的比例 44.9%；周阴线里 52.9%
周阳线从开盘向下最深跌了多少：中位数 -2.28%，超过 3% 的 39.6%

weekly["sma20"] = weekly["close"].rolling(20).mean()
weekly["up"] = (weekly["close"] > weekly["sma20"]).where(weekly["sma20"].notna())
four = four.copy()
four["sma20"] = four["close"].rolling(20).mean()
four["wk_up"] = T.align_higher(four.index, "4h", weekly[["up"]], "7D")["up"]
known = four.dropna(subset=["sma20", "wk_up"])
table = pd.crosstab(known["wk_up"].astype(bool), known["close"] > known["sma20"], normalize="index")
table.index, table.columns = ["周线在 20 周均线下", "周线在 20 周均线上"], ["4 小时线在 20 根均线下", "4 小时线在 20 根均线上"]
print(table.round(3))

             4 小时线在 20 根均线下  4 小时线在 20 根均线上
周线在 20 周均线下           0.497           0.503
周线在 20 周均线上           0.464           0.536

four["low_7d"] = four["low"].rolling(42).min().shift(1)             # 前 42 根（7 天）的最低价
four["break"] = four["close"] < four["low_7d"]
four["event"] = four["break"] & (four["break"].rolling(42).sum().shift(1) == 0)   # 7 天内第一次破位
c = four["close"]
for label, n in [("7 天", 42), ("30 天", 180)]:
    four[label] = c.shift(-n) / c - 1
rows = []
for trend, name in [(True, "周线在均线上"), (False, "周线在均线下")]:
    base = four[four["wk_up"] == trend]
    ev = base[base["event"]]
    for label in ["7 天", "30 天"]:
        e, b = ev[label].dropna(), base[label].dropna()
        t_stat = (e.mean() - b.mean()) / math.sqrt(e.var() / len(e) + b.var() / len(b))
        rows.append({"周线": name, "之后": label, "破位次数": len(e), "破位后平均": e.mean(), "破位后上涨比例": (e > 0).mean(),
                     "所有时刻平均": b.mean(), "所有时刻上涨比例": (b > 0).mean(), "t": t_stat})
print(pd.DataFrame(rows).round(4).to_string(index=False))
depth = (four["low_7d"] - four["close"]) / four["close"]
print(f"破位时收盘价在 7 天最低价下方多少：中位数 {depth[four['event']].median():.2%}")

    周线   之后  破位次数   破位后平均  破位后上涨比例  所有时刻平均  所有时刻上涨比例       t
周线在均线上  7 天    70 -0.0013   0.4857  0.0101    0.5209 -1.0317
周线在均线上 30 天    70  0.0210   0.4000  0.0535    0.5224 -1.1380
周线在均线下  7 天    57  0.0107   0.5789  0.0050    0.5324  0.4990
周线在均线下 30 天    56  0.0411   0.6429  0.0144    0.5507  1.0829
破位时收盘价在 7 天最低价下方多少：中位数 0.72%

7 天后：-5.29%
30 天后：-15.46%
之后的最低价： 57800.19 2026-07-01

def triple_screen(daily, weekly_bars, intraday=None, hold=10):
    wk = weekly_bars.copy()
    wk["sma20"] = wk["close"].rolling(20).mean()
    wk["up"] = (wk["close"] > wk["sma20"]).where(wk["sma20"].notna())
    tide = T.align_higher(daily.index, "1D", wk[["up"]], "7D")["up"]           # 第一层：周线方向
    wave = daily["close"] <= daily["close"].rolling(5).min()                   # 第二层：日线回调到 5 天最低收盘
    trades, busy_until = [], -1
    for i in range(5, len(daily) - hold - 1):
        if not wave.iloc[i] or pd.isna(tide.iloc[i]) or i <= busy_until:
            continue
        trigger = daily["high"].iloc[i]                                        # 第三层：第二天突破信号日最高价才买入
        next_day = daily.index[i + 1]
        if intraday is not None:
            bars = intraday.loc[next_day:next_day + pd.Timedelta("1D") - pd.Timedelta("1s")]
            hit = bars[bars["high"] >= trigger]
            if hit.empty:
                continue
            fill = max(trigger, hit["open"].iloc[0])                            # 跳空越过触发价时，只能按开盘价成交
        else:
            if daily["high"].iloc[i + 1] < trigger:
                continue
            fill = max(trigger, daily["open"].iloc[i + 1])
        exit_price = daily["close"].iloc[i + 1 + hold]
        trades.append({"信号日": daily.index[i], "周线向上": bool(tide.iloc[i]), "收益": exit_price / fill - 1})
        busy_until = i + 1 + hold
    return pd.DataFrame(trades)

for name, daily, intraday in [("BTC", day, four), ("SPY", spy.assign(volume=spy["volume"].fillna(0)), None)]:
    trades = triple_screen(daily, B.resample_ohlcv(daily, "W-MON"), intraday)
    base = (daily["close"].shift(-10) / daily["close"] - 1).dropna()
    print(f"{name}：任意一天买入持有 10 天，平均 {base.mean():+.2%}，上涨比例 {(base > 0).mean():.1%}")
    for label, part in [("全部信号", trades), ("周线向上", trades[trades["周线向上"]]), ("周线向下", trades[~trades["周线向上"]])]:
        r = part["收益"]
        print(f"   {label}：{len(r):3d} 笔，平均 {r.mean():+.2%}，中位数 {r.median():+.2%}，上涨比例 {(r > 0).mean():.1%}")
    up, down = trades.loc[trades["周线向上"], "收益"], trades.loc[~trades["周线向上"], "收益"]
    print(f"   周线向上 vs 向下，平均收益差异的 t = {(up.mean() - down.mean()) / math.sqrt(up.var() / len(up) + down.var() / len(down)):.2f}")

BTC：任意一天买入持有 10 天，平均 +1.54%，上涨比例 54.2%
   全部信号：127 笔，平均 +2.80%，中位数 +1.43%，上涨比例 55.9%
   周线向上： 64 笔，平均 +3.11%，中位数 +1.33%，上涨比例 57.8%
   周线向下： 63 笔，平均 +2.49%，中位数 +2.24%，上涨比例 54.0%
   周线向上 vs 向下，平均收益差异的 t = 0.33
SPY：任意一天买入持有 10 天，平均 +0.56%，上涨比例 65.0%
   全部信号：101 笔，平均 +0.74%，中位数 +0.90%，上涨比例 65.3%
   周线向上： 78 笔，平均 +0.43%，中位数 +0.85%，上涨比例 66.7%
   周线向下： 23 笔，平均 +1.81%，中位数 +2.12%，上涨比例 60.9%
   周线向上 vs 向下，平均收益差异的 t = -1.46

def align_wrong(lower_index, higher):
    """错误的对齐：用大 K 线的开始时间去匹配，会用到还没收盘的大 K 线。"""
    return higher.reindex(lower_index, method="ffill")

right = T.align_higher(four.index, "4h", weekly[cols + ["sma20", "up"]], "7D")
bad = align_wrong(four.index, weekly[cols + ["sma20", "up"]])
for name, view in [("正确对齐", right), ("错误对齐", bad)]:
    row = view.loc[t]
    print(f"{name}：周线收盘 {row['close']:,.2f}，20 周均线 {row['sma20']:,.2f}，在均线上方：{bool(row['up'])}")
dev = T.developing(four, "W-MON")
print("正在形成的周线：", dev.loc[t, cols].round(2).to_dict())

正确对齐：周线收盘 114,559.40，20 周均线 113,398.93，在均线上方：True
错误对齐：周线收盘 110,540.68，20 周均线 113,646.26，在均线上方：False
正在形成的周线： {'open': 114559.41, 'high': 116400.0, 'low': 107328.0, 'close': 107711.56}

def backtest(position, returns, periods_per_year):
    """position 是每根 K 线收盘时决定的仓位（1 持有，0 空仓），从下一根 K 线开始生效。"""
    equity = (1 + position.shift(1).fillna(0) * returns.fillna(0)).cumprod()
    growth = equity.iloc[-1]
    return growth, growth ** (periods_per_year / len(returns)) - 1

weekly["green"] = weekly["close"] > weekly["open"]
default["green"] = default["close"] > default["open"]
r4 = four["close"].pct_change()
tests = {
    "错误对齐（周一标签）": align_wrong(four.index, weekly[["green"]])["green"],
    "错误对齐（pandas 默认 W，周日标签）": align_wrong(four.index, default[["green"]])["green"],
    "正确对齐（上一根已收盘的周线）": T.align_higher(four.index, "4h", weekly[["green"]], "7D")["green"],
    "正在形成的周线": T.developing(four, "W-MON").eval("close > open"),
}
correct = tests["正确对齐（上一根已收盘的周线）"].astype(float).fillna(0)
print("规则：这一周是阳线就持有 BTC，4 小时线")
for name, signal in tests.items():
    pos = signal.astype(float).fillna(0)
    growth, cagr = backtest(pos, r4, 6 * 365)
    print(f"   {name}：×{growth:,.2f}（年化 {cagr:.1%}），和正确对齐不同的 K 线 {int((pos != correct).sum())} 根")
print(f"   一直持有：×{four['close'].iloc[-1] / four['close'].iloc[0]:.2f}")

rd = day["close"].pct_change()
print("规则：周线收盘在 20 周均线上方就持有，日线")
for name, market, bars, ppy in [("BTC", day, weekly, 365), ("SPY", spy, B.resample_ohlcv(spy.assign(volume=spy["volume"].fillna(0)), "W-MON"), 252)]:
    bars = bars.copy()
    bars["sma20"] = bars["close"].rolling(20).mean()
    bars["up"] = (bars["close"] > bars["sma20"]).where(bars["sma20"].notna())
    ret = market["close"].pct_change()
    good = T.align_higher(market.index, "1D", bars[["up"]], "7D")["up"].astype(float).fillna(0)
    leak = align_wrong(market.index, bars[["up"]])["up"].astype(float).fillna(0)
    g1, c1 = backtest(leak, ret, ppy)
    g2, c2 = backtest(good, ret, ppy)
    print(f"   {name}：错误对齐 ×{g1:.2f}（年化 {c1:.1%}）  正确对齐 ×{g2:.2f}（年化 {c2:.1%}）  "
          f"不同的日子 {int((good != leak).sum())} 天  一直持有 ×{market['close'].iloc[-1] / market['close'].iloc[0]:.2f}")

规则：这一周是阳线就持有 BTC，4 小时线
   错误对齐（周一标签）：×13,598,190.31（年化 515.5%），和正确对齐不同的 K 线 9923 根
   错误对齐（pandas 默认 W，周日标签）：×14.25（年化 34.2%），和正确对齐不同的 K 线 1210 根
   正确对齐（上一根已收盘的周线）：×7.27（年化 24.5%），和正确对齐不同的 K 线 0 根
   正在形成的周线：×50.12（年化 54.2%），和正确对齐不同的 K 线 9944 根
   一直持有：×18.06
规则：周线收盘在 20 周均线上方就持有，日线
   BTC：错误对齐 ×374.13（年化 92.5%）  正确对齐 ×7.45（年化 24.9%）  不同的日子 246 天  一直持有 ×18.34
   SPY：错误对齐 ×6.89（年化 21.4%）  正确对齐 ×2.27（年化 8.6%）  不同的日子 189 天  一直持有 ×3.52

talab/
├── talab/
│   ├── __init__.py
│   ├── data.py
│   ├── plot.py
│   ├── stats.py
│   ├── bars.py
│   └── timeframes.py     ← 新增
├── tests/
│   ├── test_data.py
│   ├── test_plot.py
│   ├── test_stats.py
│   ├── test_bars.py
│   └── test_timeframes.py  ← 新增
├── scripts/
│   └── fetch_course_data.py
├── pyproject.toml
└── requirements.txt

import glob
import math

import numpy as np
import pandas as pd
from talab import bars as B, data as D, timeframes as T

cols = ["open", "high", "low", "close"]
day = D.load_binance_klines(glob.glob("data/binance/spot/BTCUSDT/1d/*.zip"))
minute = D.load_binance_klines(sorted(glob.glob("data/binance/spot/BTCUSDT/1m/*.zip")))
D.download_binance_klines("BTCUSDT", "4h", "2017-08", "2026-08")     # 官方的 4 小时线和周线，只用来核对
D.download_binance_klines("BTCUSDT", "1w", "2017-08", "2026-06")     # 周线的月度文件目前只发布到 2026-06
official_4h = D.load_binance_klines(sorted(glob.glob("data/binance/spot/BTCUSDT/4h/*.zip")))
official_1w = D.load_binance_klines(sorted(glob.glob("data/binance/spot/BTCUSDT/1w/*.zip")))
spy = D.load_nasdaq_daily("data/nasdaq/SPY_historical.json").drop(pd.Timestamp("2026-04-20"))
print("官方 4 小时线", len(official_4h), "根，官方周线", len(official_1w), "根，最后一周", official_1w.index[-1].date())

官方 4 小时线 19794 根，官方周线 456 根，最后一周 2026-06-29

"""talab.timeframes：多周期对齐。第 7 篇。"""
from __future__ import annotations

import pandas as pd


def align_higher(lower_index: pd.DatetimeIndex, lower_period: str,
                 higher: pd.DataFrame, higher_period: str) -> pd.DataFrame:
    """把大周期的数据对齐到小周期的每一根 K 线上，只使用当时已经收盘的大周期 K 线。

    lower_index:   小周期 K 线的时间戳（开始时间）
    lower_period:  小周期的长度，比如 "4h"、"1D"
    higher:        大周期的数据，索引是每根大 K 线的开始时间
    higher_period: 大周期的长度，比如 "1D"、"7D"

    规则：小周期 K 线在它收盘的那一刻做决定；这一刻之前（含这一刻）已经收盘的大周期 K 线才能使用。
    大 K 线的收盘时间 = 开始时间 + higher_period；小 K 线的收盘时间 = 开始时间 + lower_period。
    返回的表和 lower_index 一一对应；还没有任何大 K 线收盘时为 NaN。
    """
    available = higher.copy()
    available.index = higher.index + pd.Timedelta(higher_period)          # 每根大 K 线从收盘那一刻起才可用
    decide_at = lower_index + pd.Timedelta(lower_period)                   # 每根小 K 线收盘、做决定的时刻
    out = available.reindex(decide_at, method="ffill")                     # 取这一刻之前最近一根已收盘的大 K 线
    out.index = lower_index
    return out


def developing(lower: pd.DataFrame, rule: str) -> pd.DataFrame:
    """正在形成的大周期 K 线：在每一根小 K 线收盘时，这根大 K 线「到目前为止」的样子。

    开盘 = 这个大周期第一根小 K 线的开盘价（固定不变）
    最高 / 最低 = 到目前为止的最高 / 最低
    收盘 = 当前这根小 K 线的收盘价（还会继续变）
    成交量 = 到目前为止的累计
    每个大周期最后一根小 K 线上的值，就等于这根大 K 线收盘后的最终值。
    rule 和 resample 的写法一样，比如 "1D"、"W-MON"（周一开始的一周）。
    """
    groups = pd.Grouper(freq=rule, label="left", closed="left")
    return pd.DataFrame({
        "open": lower["open"].groupby(groups).transform("first"),
        "high": lower["high"].groupby(groups).cummax(),
        "low": lower["low"].groupby(groups).cummin(),
        "close": lower["close"],
        "volume": lower["volume"].groupby(groups).cumsum(),
    }, index=lower.index)

"""talab.timeframes 的测试。"""
import numpy as np
import pandas as pd
import pytest

from talab import bars as B
from talab import timeframes as T


def daily(n=21, start="2024-01-01"):
    """从 2024-01-01（周一）开始的日线，收盘价依次是 1, 2, 3, ……"""
    idx = pd.date_range(start, periods=n, freq="1D", tz="UTC")
    close = np.arange(1, n + 1, dtype=float)
    return pd.DataFrame({"open": close - 0.5, "high": close + 1, "low": close - 1,
                         "close": close, "volume": 1.0}, index=idx)


def test_weekly_bars_start_on_monday():
    weeks = B.resample_ohlcv(daily(), "W-MON")
    assert (weeks.index.dayofweek == 0).all()
    assert weeks["close"].tolist() == [7, 14, 21]        # 每周最后一天（周日）的收盘价


def test_align_higher_by_hand():
    d = daily()
    weeks = B.resample_ohlcv(d, "W-MON")
    aligned = T.align_higher(d.index, "1D", weeks, "7D")
    # 第一周（1 月 1 日到 7 日）在 1 月 8 日 0 点收盘。1 月 7 日（周日）收盘的时刻正好是 1 月 8 日 0 点，可以用
    assert aligned["close"].iloc[:6].isna().all()          # 1 月 1 日到 6 日：还没有任何一周收盘
    assert aligned.loc["2024-01-07", "close"] == 7
    assert aligned.loc["2024-01-10", "close"] == 7          # 周三只能看到上一周
    assert aligned.loc["2024-01-14", "close"] == 14


def test_align_higher_never_uses_the_future():
    d = daily(60)
    weeks = B.resample_ohlcv(d, "W-MON")
    before = T.align_higher(d.index, "1D", weeks, "7D")
    changed = weeks.copy()
    cutoff = pd.Timestamp("2024-01-29", tz="UTC")            # 修改从这一周开始的所有周线
    changed.loc[changed.index >= cutoff, "close"] = -999
    after = T.align_higher(d.index, "1D", changed, "7D")
    # 在修改的那一周收盘（2 月 5 日 0 点）之前做的决定，都不应该受影响
    safe = d.index + pd.Timedelta("1D") < cutoff + pd.Timedelta("7D")
    pd.testing.assert_series_equal(before.loc[safe, "close"], after.loc[safe, "close"])
    assert (after.loc[~safe, "close"] == -999).any()


def test_align_4h_to_daily():
    idx = pd.date_range("2024-01-01", periods=12, freq="4h", tz="UTC")     # 两天，每天 6 根
    four = pd.DataFrame({"close": np.arange(12.0)}, index=idx)
    days = pd.DataFrame({"close": [100.0, 200.0]},
                        index=pd.DatetimeIndex(["2024-01-01", "2024-01-02"], tz="UTC"))
    aligned = T.align_higher(four.index, "4h", days, "1D")
    assert aligned["close"].iloc[:5].isna().all()            # 第一天 0 点到 20 点开始的前 5 根：日线还没收盘
    assert aligned["close"].iloc[5] == 100                   # 20:00 这根在 24:00 收盘，正好等于日线收盘时刻
    assert aligned["close"].iloc[11] == 200


def test_developing_ends_equal_to_finished_bar():
    d = daily(21)
    dev = T.developing(d, "W-MON")
    weeks = B.resample_ohlcv(d, "W-MON")
    last_of_week = dev.groupby(pd.Grouper(freq="W-MON", label="left", closed="left")).tail(1)
    assert last_of_week[["open", "high", "low", "close", "volume"]].values.tolist() == \
        weeks[["open", "high", "low", "close", "volume"]].values.tolist()
    # 周三（第 3 天）：开盘是周一的开盘价，最高是前三天的最高价，收盘是周三的收盘价
    wed = dev.loc["2024-01-03"]
    assert wed.tolist() == [0.5, 4.0, 0.0, 3.0, 3.0]


def test_developing_only_uses_the_past():
    d = daily(21)
    before = T.developing(d, "W-MON")
    changed = d.copy()
    changed.loc["2024-01-05":, ["high", "close"]] = 999          # 修改周五及以后
    after = T.developing(changed, "W-MON")
    pd.testing.assert_frame_equal(before.loc[:"2024-01-04"], after.loc[:"2024-01-04"])

pytest -q

...........................................                              [100%]
43 passed in 0.37s