第 6 篇：K 线：四个数字的有损压缩

five = minute.loc["2026-07-01 22:20":"2026-07-01 22:24", ["open", "high", "low", "close", "volume"]]
print(B.resample_ohlcv(five, "5min"))

                               open     high      low    close     volume
time                                                                     
2026-07-01 22:20:00+00:00  60819.01  61334.0  60819.0  60976.0  866.72108

cols = ["open", "high", "low", "close"]
for traded_only in [False, True]:
    rebuilt = B.resample_ohlcv(minute, "1D", traded_only=traded_only)
    both = day[cols].join(rebuilt[cols], rsuffix="_合成")
    wrong = pd.DataFrame({c: (both[c] - both[c + "_合成"]).abs() > 1e-8 for c in cols})
    bad = wrong.any(axis=1)
    print(f"traded_only={traded_only}：价格对不上的日子 {bad.sum()} 天，按字段 {wrong.sum().to_dict()}")
    print("   按年：", bad.groupby(both.index.year).sum()[lambda s: s > 0].to_dict())

traded_only=False：价格对不上的日子 43 天，按字段 {'open': 41, 'high': 1, 'low': 1, 'close': 11}
   按年： {2017: 41, 2018: 2}
traded_only=True：价格对不上的日子 16 天，按字段 {'open': 14, 'high': 1, 'low': 1, 'close': 11}
   按年： {2017: 14, 2018: 2}

                              open     high      low    close    volume
time                                                                   
2017-08-20 00:04:00+00:00  4139.98  4139.98  4139.98  4139.98  0.000000
2017-08-20 00:05:00+00:00  4139.98  4139.98  4139.98  4139.98  0.000000
2017-08-20 00:06:00+00:00  4120.98  4139.98  4120.98  4139.98  0.575365
2017-08-20 00:07:00+00:00  4139.98  4139.98  4139.98  4139.98  0.000000
官方日线 2017-08-20 开盘： 4120.98

成交量相差超过百万分之一的日子 53 天，其中相差的中位数 0.0057%，最大 4.58%（2021-04-23）

for name, t in [("A", A), ("B", Bday)]:
    print(name, B.anatomy(day.loc[[t]]).round(4).iloc[0].to_dict())

A {'body': 1399.29, 'upper_shadow': 1310.0, 'lower_shadow': 824.52, 'range': 3533.81, 'body_ratio': 0.396, 'clv': 0.6293}
B {'body': 304.81, 'upper_shadow': 251.32, 'lower_shadow': 170.6, 'range': 726.73, 'body_ratio': 0.4194, 'clv': 0.6542}

rows = {}
for k, df in {"SPY": spy, "AAPL": aapl, "BTC": day}.items():
    a = B.anatomy(df)
    rows[k] = {"实体占比中位数": a["body_ratio"].median(), "十字星(实体<10%)": (a["body_ratio"] < 0.1).mean(),
               "光头光脚(实体>90%)": (a["body_ratio"] > 0.9).mean(), "收盘位置中位数": a["clv"].median()}
print(pd.DataFrame(rows).round(3))

                SPY   AAPL    BTC
实体占比中位数       0.473  0.458  0.413
十字星(实体<10%)   0.105  0.109  0.118
光头光脚(实体>90%)  0.038  0.030  0.018
收盘位置中位数       0.592  0.544  0.549

for k, df in {"SPY": spy, "AAPL": aapl, "BTC": day}.items():
    ret = df["close"].pct_change()
    green_down = ((df["close"] > df["open"]) & (ret < 0)).sum()
    red_up = ((df["close"] < df["open"]) & (ret > 0)).sum()
    gap = df["open"] / df["close"].shift() - 1
    print(f"{k:4s} 阳线但下跌 {green_down} 天，阴线但上涨 {red_up} 天，占 {(green_down + red_up) / (len(df) - 1):.1%}；"
          f"跳空超过 0.5% 的日子 {(gap.abs() > 0.005).mean():.1%}")

SPY  阳线但下跌 240 天，阴线但上涨 267 天，占 20.2%；跳空超过 0.5% 的日子 29.1%
AAPL 阳线但下跌 220 天，阴线但上涨 213 天，占 17.2%；跳空超过 0.5% 的日子 44.9%
BTC  阳线但下跌 2 天，阴线但上涨 2 天，占 0.1%；跳空超过 0.5% 的日子 0.4%

              open    high     low   close       volume
date                                                   
2025-04-04  523.67  525.87  505.06  505.28  217965100.0
2025-04-07  489.19  523.17  481.80  504.38  256611400.0

for k, df in {"SPY": spy, "AAPL": aapl, "BTC": day}.items():
    overnight = np.log(df["open"] / df["close"].shift()).dropna()        # 昨天收盘 → 今天开盘
    intraday = np.log(df["close"] / df["open"]).iloc[1:]                 # 今天开盘 → 今天收盘
    print(f"{k:4s} 隔夜累计 ×{math.exp(overnight.sum()):.2f}   日内累计 ×{math.exp(intraday.sum()):.2f}   "
          f"合计 ×{math.exp(overnight.sum() + intraday.sum()):.2f}")

SPY  隔夜累计 ×2.40   日内累计 ×1.47   合计 ×3.52
AAPL 隔夜累计 ×1.04   日内累计 ×11.06   合计 ×11.47
BTC  隔夜累计 ×0.98   日内累计 ×18.74   合计 ×18.34

ex = B.intraday_extremes(minute)
d = day.join(ex)
up, down = d["close"] > d["open"], d["close"] < d["open"]
print(f"全部日子：最高价先出现 {d['high_first'].mean():.1%}")
print(f"阳线：最高价先出现 {d.loc[up, 'high_first'].mean():.1%}，最低价先出现 {1 - d.loc[up, 'high_first'].mean():.1%}")
print(f"阴线：最高价先出现 {d.loc[down, 'high_first'].mean():.1%}")

全部日子：最高价先出现 50.3%
阳线：最高价先出现 13.4%，最低价先出现 86.6%
阴线：最高价先出现 89.1%

theory = [(2 / math.pi) * (math.asin(math.sqrt((h + 1) / 24)) - math.asin(math.sqrt(h / 24))) for h in range(24)]
hours = pd.DataFrame({
    "最高价": ex["time_of_high"].dt.hour.value_counts(normalize=True).sort_index(),
    "最低价": ex["time_of_low"].dt.hour.value_counts(normalize=True).sort_index(),
    "随机游走理论": theory,
})
print((hours * 100).round(1).T.to_string())

          0    1    2    3    4    5    6    7    8    9    10   11   12   13   14   15   16   17   18   19   20   21   22    23
最高价     15.7  6.0  3.5  2.7  2.4  2.2  2.4  2.4  2.7  2.3  2.2  2.0  3.9  4.1  4.5  4.4  4.9  3.2  3.5  3.4  4.5  4.1  5.0   8.0
最低价     15.6  5.8  3.8  3.2  3.2  2.6  2.4  2.4  2.5  2.4  2.5  2.6  3.3  4.2  5.2  4.5  4.6  3.5  3.6  3.3  4.8  3.6  4.0   6.4
随机游走理论  13.1  5.6  4.4  3.8  3.4  3.2  3.0  2.9  2.8  2.7  2.7  2.7  2.7  2.7  2.7  2.8  2.9  3.0  3.2  3.4  3.8  4.4  5.6  13.1

for k in [0.02, 0.03, 0.05]:
    truth = B.first_touch(minute, k, k)
    hit_up = day["high"] >= day["open"] * (1 + k)
    hit_down = day["low"] <= day["open"] * (1 - k)
    ambiguous = hit_up & hit_down
    only_up, only_down, neither = hit_up & ~hit_down, hit_down & ~hit_up, ~hit_up & ~hit_down
    base = k * only_up.sum() - k * only_down.sum() + (day["close"] / day["open"] - 1)[neither].sum()
    t = truth[ambiguous]
    guess = np.where(day.loc[ambiguous, "close"] >= day.loc[ambiguous, "open"], "down", "up")   # 阳线假设先低后高
    n = ambiguous.sum()
    print(f"±{k:.0%}：{len(day)} 天里，只碰到止盈 {only_up.sum()} 天，只碰到止损 {only_down.sum()} 天，都没碰到 {neither.sum()} 天，两个都碰到 {n} 天")
    print(f"   两个都碰到的 {n} 天里，1 分钟数据显示先碰到止盈 {(t == 'up').sum()} 天，先碰到止损 {(t == 'down').sum()} 天，同一分钟 {(t == 'same_bar').sum()} 天")
    print(f"   收益率之和：假设先止盈 {(base + k * n):+.1%}   假设先止损 {(base - k * n):+.1%}   "
          f"按阴阳线猜 {(base + k * (guess == 'up').sum() - k * (guess == 'down').sum()):+.1%}（猜对 {(guess == t.values).mean():.1%}）   "
          f"真实 {(base + k * (t == 'up').sum() - k * (t == 'down').sum()):+.1%}")

±2%：3302 天里，只碰到止盈 959 天，只碰到止损 1001 天，都没碰到 951 天，两个都碰到 391 天
   两个都碰到的 391 天里，1 分钟数据显示先碰到止盈 191 天，先碰到止损 200 天，同一分钟 0 天
   收益率之和：假设先止盈 +745.0%   假设先止损 -819.0%   按阴阳线猜 -83.0%（猜对 74.2%）   真实 -55.0%
±3%：3302 天里，只碰到止盈 722 天，只碰到止损 768 天，都没碰到 1666 天，两个都碰到 146 天
   两个都碰到的 146 天里，1 分钟数据显示先碰到止盈 77 天，先碰到止损 69 天，同一分钟 0 天
   收益率之和：假设先止盈 +422.5%   假设先止损 -453.5%   按阴阳线猜 -87.5%（猜对 75.3%）   真实 +8.5%
±5%：3302 天里，只碰到止盈 378 天，只碰到止损 384 天，都没碰到 2508 天，两个都碰到 32 天
   两个都碰到的 32 天里，1 分钟数据显示先碰到止盈 14 天，先碰到止损 18 天，同一分钟 0 天
   收益率之和：假设先止盈 +315.6%   假设先止损 -4.4%   按阴阳线猜 +75.6%（猜对 68.8%）   真实 +135.6%

d["next"] = d["close"].pct_change().shift(-1)
for name, mask in [("阳线", up), ("阴线", down)]:
    a = d.loc[mask & (d["high_first"] == True), "next"].dropna()
    b = d.loc[mask & (d["high_first"] == False), "next"].dropna()
    t_stat = (a.mean() - b.mean()) / math.sqrt(a.var() / len(a) + b.var() / len(b))
    print(f"{name}：先高后低 {len(a)} 天，次日平均 {a.mean():+.3%}；先低后高 {len(b)} 天，次日平均 {b.mean():+.3%}；t = {t_stat:.2f}")

阳线：先高后低 226 天，次日平均 +0.276%；先低后高 1463 天，次日平均 +0.090%；t = 0.75
阴线：先高后低 1436 天，次日平均 +0.224%；先低后高 175 天，次日平均 -0.109%；t = 1.33

最后两小时涨跌 vs 次日涨跌：相关系数 -0.056，界限 ±0.034
   2017–2019：-0.077（867 天）
   2020–2020：-0.197（366 天）
   2021–2023：+0.053（1095 天）
   2024–2026：-0.060（973 天）
   去掉 2020 年：-0.027

A 2026-07-01 {'1 天后': '+2.56%', '3 天后': '+5.20%', '7 天后': '+3.78%', '10 天后': '+6.32%'}

B 2019-07-09 {'1 天后': '-3.47%', '3 天后': '-6.27%', '7 天后': '-24.74%', '10 天后': '-16.26%'}

for name, t in [("A", A), ("B", Bday)]:
    m = minute.loc[t]
    print(name, {f"±{k:.0%}": B.first_touch(m, k, k).iloc[0] for k in [0.01, 0.03]})

A {'±1%': 'down', '±3%': 'up'}
B {'±1%': 'up', '±3%': 'up'}

ha = B.heikin_ashi(spy)
print(pd.concat([spy[cols], ha.add_prefix("ha_")], axis=1).loc["2025-04-02":"2025-04-10"].round(4))

              open     high     low   close   ha_open   ha_high  ha_low  ha_close
date                                                                             
2025-04-02  555.05  567.420  554.81  564.52  559.2092  567.4200  554.81  560.4500
2025-04-03  545.11  547.970  536.70  536.70  559.8296  559.8296  536.70  541.6200
2025-04-04  523.67  525.870  505.06  505.28  550.7248  550.7248  505.06  514.9700
2025-04-07  489.19  523.170  481.80  504.38  532.8474  532.8474  481.80  499.6350
2025-04-08  521.86  524.980  489.16  496.48  516.2412  524.9800  489.16  508.1200
2025-04-09  493.44  548.620  493.05  548.62  512.1806  548.6200  493.05  520.9325
2025-04-10  532.17  533.495  509.32  524.58  516.5565  533.4950  509.32  524.8912

从不同的起点开始算，HA 开盘的差距： ['1.0331%', '0.5162%', '0.2604%', '0.1301%', '0.0642%', '0.0318%', '0.0156%', '0.0075%']

for k, df in {"SPY": spy, "AAPL": aapl, "BTC": day}.items():
    h = B.heikin_ashi(df)
    outside = ((h["open"] > df["high"]) | (h["open"] < df["low"])).mean()
    close_gap = (h["close"] / df["close"] - 1).abs()
    print(f"{k:4s} HA 开盘在真实最高最低之外 {outside:.1%}；HA 收盘偏离真实收盘 中位数 {close_gap.median():.2%} 最大 {close_gap.max():.1%}；"
          f"同色连续根数 平均 真实 {runs(np.sign(df['close'] - df['open'])).mean():.2f} / HA {runs(np.sign(h['close'] - h['open'])).mean():.2f}")

SPY  HA 开盘在真实最高最低之外 57.7%；HA 收盘偏离真实收盘 中位数 0.21% 最大 5.0%；同色连续根数 平均 真实 2.00 / HA 3.77
AAPL HA 开盘在真实最高最低之外 57.6%；HA 收盘偏离真实收盘 中位数 0.40% 最大 6.6%；同色连续根数 平均 真实 1.98 / HA 4.00
BTC  HA 开盘在真实最高最低之外 46.5%；HA 收盘偏离真实收盘 中位数 0.76% 最大 30.8%；同色连续根数 平均 真实 1.88 / HA 4.24

for k, df in {"SPY": spy, "AAPL": aapl, "BTC": day}.items():
    years = len(df) / (365 if k == "BTC" else 252)
    n, f, fc, r, rc, no, noc, bh = ha_backtest(df, years)
    print(f"{k:4s} {n} 笔：按 HA 收盘价 ×{f:,.2f}（年化 {fc:.1%}）  按真实收盘价 ×{r:.2f}（{rc:.1%}）  "
          f"按次日开盘价 ×{no:.2f}（{noc:.1%}）  一直持有 ×{bh:.2f}")

SPY  333 笔：按 HA 收盘价 ×8.97（年化 24.6%）  按真实收盘价 ×2.01（7.2%）  按次日开盘价 ×1.80（6.1%）  一直持有 ×3.52
AAPL 314 笔：按 HA 收盘价 ×59.59（年化 50.7%）  按真实收盘价 ×4.43（16.1%）  按次日开盘价 ×1.46（3.8%）  一直持有 ×11.47
BTC  389 笔：按 HA 收盘价 ×131,537.90（年化 268.0%）  按真实收盘价 ×32.48（46.9%）  按次日开盘价 ×32.57（47.0%）  一直持有 ×18.34

真实收盘 548.62  HA 收盘 520.9325  止损放在 HA 最低 493.05
以为的风险 27.8825（5.35%）  实际的风险 55.5700（10.13%）

SPY  实际风险 ÷ 以为的风险：大于 1.5 倍 16.5%，小于 1/1.5 24.0%
AAPL 实际风险 ÷ 以为的风险：大于 1.5 倍 17.4%，小于 1/1.5 26.0%
BTC  实际风险 ÷ 以为的风险：大于 1.5 倍 17.7%，小于 1/1.5 24.3%

closes = day["close"].loc["2026-06-24":"2026-07-12"]
print(B.renko(closes, 1000))

                  formed_at      open     close  direction
0 2026-06-25 00:00:00+00:00  61077.99  60077.99         -1
1 2026-06-30 00:00:00+00:00  60077.99  59077.99         -1
2 2026-07-02 00:00:00+00:00  60077.99  61077.99          1
3 2026-07-03 00:00:00+00:00  61077.99  62077.99          1
4 2026-07-04 00:00:00+00:00  62077.99  63077.99          1
5 2026-07-10 00:00:00+00:00  63077.99  64077.99          1

daily_bricks = B.renko(day["close"], 0.05, log=True)
minute_bricks = B.renko(minute["close"], 0.05, log=True)
for name, bricks in [("日线收盘价", daily_bricks), ("1 分钟收盘价", minute_bricks)]:
    reversals = (bricks["direction"] != bricks["direction"].shift()).sum() - 1
    print(f"用 {name}：{len(bricks)} 块砖，{reversals} 次反向")

用 日线收盘价：541 块砖，123 次反向
用 1 分钟收盘价：972 块砖，300 次反向
两块砖之间最长隔了 85 天： 2025-07-13 → 2025-10-06
一天里最多画了 10 块砖： 2020-03-12
起点从 2017-08-17 换成 2017-08-18： 541 块 → 524 块

形成砖块的那天，真实收盘价比最后一块砖的收盘价多走了：中位数 1.42%，最大 4.95%
砖块转为向上就买、转为向下就卖（只做多），62 笔：按砖块价格 ×281.81（年化 86.6%）  按真实收盘价 ×14.71（年化 34.6%）

talab/
├── talab/
│   ├── __init__.py
│   ├── data.py
│   ├── plot.py
│   ├── stats.py
│   └── bars.py          ← 新增
├── tests/
│   ├── test_data.py
│   ├── test_plot.py
│   ├── test_stats.py
│   └── test_bars.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

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")))
spy = D.load_nasdaq_daily("data/nasdaq/SPY_historical.json").drop(pd.Timestamp("2026-04-20"))
aapl = D.load_nasdaq_daily("data/nasdaq/AAPL_historical.json")
print("BTC 日线", len(day), "根，1 分钟线", len(minute), "根")
A, Bday = "2026-07-01", "2019-07-09"
cols = ["open", "high", "low", "close"]

BTC 日线 3302 根，1 分钟线 4746079 根

"""talab.bars：K 线的构造与拆解。第 6 篇。"""
from __future__ import annotations

import math

import numpy as np
import pandas as pd

OHLCV = {"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"}
SUMMABLE = ["quote_volume", "trades", "taker_buy_base", "taker_buy_quote"]


# ---------------------------------------------------------------------------
# 一、合成：小周期 → 大周期
# ---------------------------------------------------------------------------

def resample_ohlcv(df: pd.DataFrame, rule: str, traded_only: bool = False) -> pd.DataFrame:
    """把小周期 K 线合成大周期 K 线。

    开盘取第一根的开盘价，最高取最大值，最低取最小值，收盘取最后一根的收盘价，量类字段求和。
    每根大 K 线的时间戳是它的开始时间，包含开始时刻、不包含结束时刻。
    traded_only=True 时先去掉成交量为 0 的占位 K 线，否则占位 K 线的价格会被当成真实成交价。
    一根成交也没有的时间段不会出现在结果里。
    """
    src = df[df["volume"] > 0] if traded_only else df
    rules = dict(OHLCV)
    rules.update({c: "sum" for c in SUMMABLE if c in src.columns})
    out = src.resample(rule, label="left", closed="left").agg(rules)
    return out.dropna(subset=["open"])


# ---------------------------------------------------------------------------
# 二、拆解：实体和影线
# ---------------------------------------------------------------------------

def anatomy(df: pd.DataFrame) -> pd.DataFrame:
    """每根 K 线的各个部分。

    body          实体：收盘 − 开盘（带符号，正数是阳线）
    upper_shadow  上影线：最高 − max(开盘, 收盘)
    lower_shadow  下影线：min(开盘, 收盘) − 最低
    range         全长：最高 − 最低
    body_ratio    实体占全长的比例，0～1
    clv           收盘位置：(收盘 − 最低) ÷ 全长，0 表示收在最低，1 表示收在最高
    """
    o, h, l, c = df["open"], df["high"], df["low"], df["close"]
    rng = h - l
    safe = rng.where(rng > 0)                                 # 全长为 0 时比例没有意义，记为 NaN
    return pd.DataFrame({
        "body": c - o,
        "upper_shadow": h - np.maximum(o, c),
        "lower_shadow": np.minimum(o, c) - l,
        "range": rng,
        "body_ratio": (c - o).abs() / safe,
        "clv": (c - l) / safe,
    }, index=df.index)


# ---------------------------------------------------------------------------
# 三、K 线丢掉的路径：用小周期找回来
# ---------------------------------------------------------------------------

def intraday_extremes(minute: pd.DataFrame, rule: str = "1D") -> pd.DataFrame:
    """每个大周期里，最高价和最低价分别出现在哪根小 K 线。

    同一个最高价出现多次时取第一次。high_first 为 True 表示最高价先于最低价出现；
    两者出现在同一根小 K 线里时，小 K 线本身也分不清先后，记为 NA。
    """
    period = minute.index.floor(rule)
    t_high = minute["high"].groupby(period).idxmax()
    t_low = minute["low"].groupby(period).idxmin()
    out = pd.DataFrame({"time_of_high": t_high, "time_of_low": t_low})
    out["high_first"] = (out["time_of_high"] < out["time_of_low"]).astype("boolean")
    out.loc[out["time_of_high"] == out["time_of_low"], "high_first"] = pd.NA
    return out


def first_touch(minute: pd.DataFrame, up: float, down: float, rule: str = "1D") -> pd.Series:
    """以每个大周期的开盘价为基准，价格先碰到 +up 还是 −down（都是比例，比如 0.03）。

    返回每个周期的结果："up" 先碰到上方，"down" 先碰到下方，"none" 两边都没碰到，
    "same_bar" 两边在同一根小 K 线里都碰到了（小 K 线也分不清先后）。
    """
    period = minute.index.floor(rule)
    base = minute["open"].groupby(period).transform("first")          # 每个周期的开盘价
    hit_up = minute["high"] >= base * (1 + up)
    hit_down = minute["low"] <= base * (1 - down)
    times = minute.index.to_series(index=minute.index)
    first_up = times[hit_up].groupby(period[hit_up.to_numpy()]).first()      # 第一次碰到上方的时间
    first_down = times[hit_down].groupby(period[hit_down.to_numpy()]).first()
    days = pd.Index(period.unique())
    fu, fd = first_up.reindex(days), first_down.reindex(days)
    result = pd.Series("none", index=days, dtype=object)
    result[fu.notna() & (fd.isna() | (fu < fd))] = "up"
    result[fd.notna() & (fu.isna() | (fd < fu))] = "down"
    result[fu.notna() & fd.notna() & (fu == fd)] = "same_bar"
    return result


# ---------------------------------------------------------------------------
# 四、Heikin-Ashi：平均出来的 K 线
# ---------------------------------------------------------------------------

def heikin_ashi(df: pd.DataFrame) -> pd.DataFrame:
    """Heikin-Ashi K 线。

    HA 收盘 = (开 + 高 + 低 + 收) ÷ 4
    HA 开盘 = (前一根 HA 开盘 + 前一根 HA 收盘) ÷ 2；第一根用 (开 + 收) ÷ 2
    HA 最高 = max(最高, HA 开盘, HA 收盘)
    HA 最低 = min(最低, HA 开盘, HA 收盘)
    ⚠️ 这些价格大多不是真实成交过的价格。
    """
    o, h, l, c = (df[k].to_numpy(dtype=float) for k in ["open", "high", "low", "close"])
    ha_close = (o + h + l + c) / 4
    ha_open = np.empty(len(df))
    ha_open[0] = (o[0] + c[0]) / 2
    for i in range(1, len(df)):
        ha_open[i] = (ha_open[i - 1] + ha_close[i - 1]) / 2
    return pd.DataFrame({
        "open": ha_open,
        "high": np.maximum.reduce([h, ha_open, ha_close]),
        "low": np.minimum.reduce([l, ha_open, ha_close]),
        "close": ha_close,
    }, index=df.index)


# ---------------------------------------------------------------------------
# 五、Renko：去掉时间轴的砖块图
# ---------------------------------------------------------------------------

def renko(close: pd.Series, brick: float, log: bool = False) -> pd.DataFrame:
    """用收盘价构造 Renko 砖块。

    brick: 砖块大小。log=False 时是价格单位（比如 1000 美元）；
           log=True 时是比例（比如 0.05 表示每块 5%），砖块在对数坐标上等高。
    规则：第一个价格是起点。顺着当前方向，每走满一块砖就画一块；
         要反向，价格必须越过当前这块砖的另一端再走满一块砖（也就是离最后一块砖的收盘两块砖远）。
    返回每块砖：formed_at（形成这块砖的那根 K 线的时间）、open、close、direction（+1 或 −1）。
    """
    step = math.log(1 + brick) if log else brick
    xs = np.log(close.to_numpy(dtype=float)) if log else close.to_numpy(dtype=float)
    rows = []
    lo = hi = xs[0]                  # 最后一块砖的下沿和上沿；还没有砖时两者都是起点
    for t, x in zip(close.index, xs):
        # 价格比最后一块砖的上沿高出一块砖：画一块向上的砖。
        # 如果最后一块砖是向下的，它的上沿就是它的开盘价，所以这正好是「离收盘两块砖」的反向条件
        while x >= hi + step:
            rows.append((t, hi, hi + step, 1))
            lo, hi = hi, hi + step
        # 价格比最后一块砖的下沿低出一块砖：画一块向下的砖（道理同上）
        while x <= lo - step:
            rows.append((t, lo, lo - step, -1))
            lo, hi = lo - step, lo
    out = pd.DataFrame(rows, columns=["formed_at", "open", "close", "direction"])
    if log:
        out[["open", "close"]] = np.exp(out[["open", "close"]])
    return out

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

from talab import bars as B


def minutes(rows, start="2024-01-01 00:00"):
    idx = pd.date_range(start, periods=len(rows), freq="1min", tz="UTC")
    return pd.DataFrame(rows, columns=["open", "high", "low", "close", "volume"], index=idx, dtype=float)


def test_resample_ohlcv_by_hand():
    m = minutes([
        [10, 12, 9, 11, 1],       # 00:00
        [11, 15, 11, 14, 2],      # 00:01  最高价 15
        [14, 14, 8, 9, 3],        # 00:02  最低价 8
        [9, 10, 9, 10, 4],        # 00:03
        [10, 11, 10, 11, 5],      # 00:04  收盘 11
        [11, 13, 11, 12, 6],      # 00:05  属于下一根 5 分钟 K 线
    ])
    out = B.resample_ohlcv(m, "5min")
    assert len(out) == 2
    assert out.iloc[0].tolist() == [10, 15, 8, 11, 15]
    assert out.index[1] == pd.Timestamp("2024-01-01 00:05", tz="UTC")   # 时间戳是开始时间


def test_resample_traded_only_ignores_placeholder_bars():
    m = minutes([
        [100, 100, 100, 100, 0],  # 没有成交的占位 K 线，价格沿用上一分钟的收盘价
        [98, 101, 97, 99, 2],     # 第一笔真实成交在这里
    ])
    assert B.resample_ohlcv(m, "1D").iloc[0]["open"] == 100
    assert B.resample_ohlcv(m, "1D", traded_only=True).iloc[0]["open"] == 98


def test_anatomy_by_hand():
    df = minutes([[100, 110, 95, 104, 1], [50, 50, 50, 50, 1]])
    a = B.anatomy(df)
    first = a.iloc[0]
    assert first["body"] == 4
    assert first["upper_shadow"] == 6            # 110 − 104
    assert first["lower_shadow"] == 5            # 100 − 95
    assert first["range"] == 15
    assert first["body_ratio"] == pytest.approx(4 / 15)
    assert first["clv"] == pytest.approx(9 / 15)
    assert np.isnan(a.iloc[1]["clv"])            # 四价相同，全长为 0


def test_intraday_extremes_order():
    day1 = [[10, 10, 9, 9, 1], [9, 9, 5, 6, 1], [6, 12, 6, 12, 1]]        # 先低后高
    day2 = [[10, 20, 10, 18, 1], [18, 18, 3, 4, 1], [4, 6, 4, 5, 1]]      # 先高后低
    day3 = [[10, 30, 1, 10, 1], [10, 11, 9, 10, 1], [10, 11, 9, 10, 1]]   # 高低在同一分钟
    m = pd.concat([minutes(day1, "2024-01-01"), minutes(day2, "2024-01-02"), minutes(day3, "2024-01-03")])
    ex = B.intraday_extremes(m)
    assert ex["high_first"].iloc[0] == False
    assert ex["high_first"].iloc[1] == True
    assert pd.isna(ex["high_first"].iloc[2])
    assert ex["time_of_low"].iloc[0] == pd.Timestamp("2024-01-01 00:01", tz="UTC")


def test_first_touch_all_outcomes():
    up = [[100, 101, 99, 100, 1], [100, 104, 100, 102, 1], [102, 102, 96, 97, 1]]    # 先 +3%，后 −3%
    down = [[100, 100, 96, 97, 1], [97, 104, 97, 103, 1], [103, 103, 103, 103, 1]]  # 先 −3%，后 +3%
    none = [[100, 102, 98, 100, 1]] * 3
    both = [[100, 104, 96, 100, 1]] * 3                                             # 同一分钟两边都碰到
    m = pd.concat([minutes(up, "2024-01-01"), minutes(down, "2024-01-02"),
                   minutes(none, "2024-01-03"), minutes(both, "2024-01-04")])
    assert B.first_touch(m, 0.03, 0.03).tolist() == ["up", "down", "none", "same_bar"]


def test_heikin_ashi_by_hand():
    df = minutes([[10, 14, 8, 12, 1], [12, 16, 11, 15, 1], [15, 15, 9, 10, 1]])
    ha = B.heikin_ashi(df)
    # 第一根：HA 开盘 = (10 + 12) ÷ 2 = 11，HA 收盘 = (10 + 14 + 8 + 12) ÷ 4 = 11
    assert ha.iloc[0].tolist() == [11, 14, 8, 11]
    # 第二根：HA 开盘 = (11 + 11) ÷ 2 = 11，HA 收盘 = (12 + 16 + 11 + 15) ÷ 4 = 13.5
    assert ha.iloc[1].tolist() == [11, 16, 11, 13.5]
    # 第三根：HA 开盘 = (11 + 13.5) ÷ 2 = 12.25，HA 收盘 = (15 + 15 + 9 + 10) ÷ 4 = 12.25
    assert ha.iloc[2].tolist() == [12.25, 15, 9, 12.25]


def test_heikin_ashi_range_always_covers_real_range():
    rng = np.random.default_rng(3)
    close = 100 * np.exp(np.cumsum(rng.normal(0, 0.02, 500)))
    open_ = np.r_[100, close[:-1]]
    high = np.maximum(open_, close) * (1 + rng.uniform(0, 0.01, 500))
    low = np.minimum(open_, close) * (1 - rng.uniform(0, 0.01, 500))
    df = pd.DataFrame({"open": open_, "high": high, "low": low, "close": close})
    ha = B.heikin_ashi(df)
    assert (ha["high"] >= df["high"]).all() and (ha["low"] <= df["low"]).all()


def test_renko_by_hand():
    close = pd.Series([100, 105, 111, 121, 115, 99, 100], dtype=float)
    bricks = B.renko(close, 10)
    # 111 → 画 100～110；121 → 画 110～120；115 不够；99 跌破 110 − 10 = 100，反向画 110～100
    assert bricks[["open", "close", "direction"]].values.tolist() == [[100, 110, 1], [110, 120, 1], [110, 100, -1]]
    assert bricks["formed_at"].tolist() == [2, 3, 5]


def test_renko_needs_two_bricks_to_reverse():
    close = pd.Series([100, 110, 101, 100.5, 90], dtype=float)
    bricks = B.renko(close, 10)
    # 110 画出 100～110 的砖。跌到 101、100.5 都不反向：反向要跌到 110 − 2 × 10 = 90
    assert bricks["direction"].tolist() == [1, -1]
    assert bricks["formed_at"].tolist() == [1, 4]


def test_renko_log_bricks_are_equal_ratios():
    close = pd.Series([100, 122, 99], dtype=float)
    bricks = B.renko(close, 0.10, log=True)
    assert bricks["close"].iloc[0] == pytest.approx(110)
    assert bricks["close"].iloc[1] == pytest.approx(121)
    assert bricks["direction"].tolist() == [1, 1, -1]          # 跌到 99，低于 110 ÷ 1.1 = 100，反向一块

pytest -q

.....................................                                    [100%]
37 passed in 0.40s