第 4 篇：实验环境与第一张 K 线图

python3 --version

python3 -m venv .venv

source .venv/bin/activate

talab/
├── pyproject.toml              # 项目的描述：名字、依赖、测试设置
├── requirements.txt            # 精确的依赖版本
├── .gitignore                  # 不纳入版本管理的文件
├── talab/                      # 代码本体
│   ├── __init__.py
│   ├── data.py                 # 第 3 篇：下载、加载、调整、体检
│   └── plot.py                 # 第 4 篇：K 线绘图
├── tests/                      # 测试
│   ├── test_data.py
│   └── test_plot.py
├── scripts/
│   └── fetch_course_data.py    # 下载全部课程数据
├── data/                       # 下载的数据（不纳入版本管理）
└── figures/                    # 生成的图（不纳入版本管理）

[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"

[project]
name = "talab"
version = "0.1.0"
description = "交易技术分析课程的实验工具库"
requires-python = ">=3.11"
dependencies = [
    "pandas>=3.0",
    "numpy>=2.0",
    "matplotlib>=3.9",
    "exchange_calendars>=4.5",
]

[project.optional-dependencies]
dev = ["pytest>=8"]

[tool.setuptools]
packages = ["talab"]

[tool.pytest.ini_options]
testpaths = ["tests"]
filterwarnings = ["ignore:Glyph .* missing from font"]    # 测试环境里没有配置中文字体，不影响结果

# 正文里所有代码的实际运行环境（Python 3.14.7）
pandas==3.0.5
numpy==2.5.3
matplotlib==3.11.2
exchange_calendars==4.13.2
pytest==9.1.1

pip install -r requirements.txt

pip install -e .

"""下载这门课用到的全部基础数据，并打印复现指纹。

在项目根目录运行：python scripts/fetch_course_data.py
"""
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import pandas as pd

from talab import data as D

CRYPTO_START, CRYPTO_END = "2017-08", "2026-08"          # 加密数据：按月，包含两端
US_START, US_END = "2016-01-01", "2026-09-15"            # 美股数据：Nasdaq 接口最多回溯约 10 年
US_SYMBOLS = {"SPY": "etf", "AAPL": "stocks", "TSLA": "stocks"}


def fetch_binance(interval: str) -> None:
    months = [str(m) for m in pd.period_range(CRYPTO_START, CRYPTO_END, freq="M")]
    # 按月并行下载；每个月都会校验官方 SHA-256，已经下载且校验通过的会跳过
    with ThreadPoolExecutor(max_workers=8) as pool:
        list(pool.map(lambda m: D.download_binance_klines("BTCUSDT", interval, m, m), months))


def main() -> None:
    for interval in ["1d", "1m"]:
        print(f"下载 BTCUSDT {interval} ……", flush=True)
        fetch_binance(interval)
    for symbol, assetclass in US_SYMBOLS.items():
        print(f"下载 {symbol} 日线 ……", flush=True)
        D.download_nasdaq(symbol, "historical", assetclass=assetclass, start=US_START, end=US_END)
    D.download_nasdaq("AAPL", "dividends")

    print("\n复现指纹：")
    for interval in ["1d", "1m"]:
        df = D.load_binance_klines(Path("data/binance/spot/BTCUSDT", interval).glob("*.zip"))
        print(f"BTCUSDT {interval}:", D.fingerprint(df))
    for symbol in US_SYMBOLS:
        df = D.load_nasdaq_daily(f"data/nasdaq/{symbol}_historical.json")
        print(f"{symbol}:", D.fingerprint(df))


if __name__ == "__main__":
    main()

python scripts/fetch_course_data.py

复现指纹：
BTCUSDT 1d: {'行数': 3302, '起点': '2017-08-17 00:00:00+00:00', '终点': '2026-08-31 00:00:00+00:00', '收盘价之和': 127457855.94}
BTCUSDT 1m: {'行数': 4746079, '起点': '2017-08-17 04:00:00+00:00', '终点': '2026-08-31 23:59:00+00:00', '收盘价之和': 183332267178.49}
SPY: {'行数': 2513, '起点': '2016-09-15 00:00:00', '终点': '2026-09-15 00:00:00', '收盘价之和': 1036175.29}
AAPL: {'行数': 2513, '起点': '2016-09-15 00:00:00', '终点': '2026-09-15 00:00:00', '收盘价之和': 338953.22}
TSLA: {'行数': 2513, '起点': '2016-09-15 00:00:00', '终点': '2026-09-15 00:00:00', '收盘价之和': 442548.68}

"""talab.data 的测试。全部使用手工构造的小数据，不需要联网，也不需要下载任何文件。"""
import io
import zipfile

import numpy as np
import pandas as pd
import pytest

from talab import data as D


def make_bars(n=5, start="2024-01-01", freq="1min"):
    idx = pd.date_range(start, periods=n, freq=freq, tz="UTC")
    return pd.DataFrame({"open": 100.0, "high": 101.0, "low": 99.0, "close": 100.5, "volume": 10.0}, index=idx)


# ---------- 加载 ----------

def write_kline_zip(path, rows):
    buf = io.StringIO()
    pd.DataFrame(rows).to_csv(buf, header=False, index=False)
    with zipfile.ZipFile(path, "w") as z:
        z.writestr(path.stem + ".csv", buf.getvalue())


def kline_row(open_time):
    return [open_time, "1", "2", "0.5", "1.5", "10", open_time + 59_999, "15", 3, "4", "6", "0"]


def test_load_handles_millisecond_and_microsecond_timestamps(tmp_path):
    ms = 1735689540000          # 2024-12-31 23:59:00 UTC，毫秒
    us = 1735689600000000       # 2025-01-01 00:00:00 UTC，微秒
    write_kline_zip(tmp_path / "a-2024-12.zip", [kline_row(ms)])
    write_kline_zip(tmp_path / "a-2025-01.zip", [kline_row(us)])
    df = D.load_binance_klines(tmp_path.glob("*.zip"))
    assert list(df.index) == [pd.Timestamp("2024-12-31 23:59", tz="UTC"), pd.Timestamp("2025-01-01 00:00", tz="UTC")]


def test_load_raises_when_no_files(tmp_path):
    with pytest.raises(ValueError):
        D.load_binance_klines(tmp_path.glob("*.zip"))


# ---------- 价格调整 ----------

def daily(prices, start="2020-08-26"):
    idx = pd.bdate_range(start, periods=len(prices))
    p = pd.Series(prices, index=idx, dtype=float)
    return pd.DataFrame({"open": p, "high": p, "low": p, "close": p, "volume": 1000.0})


def test_split_roundtrip_restores_original():
    raw = daily([500.0, 510.0, 130.0, 131.0])                 # 第三天起 1 拆 4
    splits = [(raw.index[2].strftime("%Y-%m-%d"), 4)]
    adj = D.adjust_splits(raw, splits)
    assert adj["close"].tolist() == [125.0, 127.5, 130.0, 131.0]
    assert adj["volume"].tolist() == [4000.0, 4000.0, 1000.0, 1000.0]
    pd.testing.assert_frame_equal(D.unadjust_splits(adj, splits), raw)


def test_dividend_factor_matches_hand_calculation():
    # 第 3 篇练习 2：3 月 9 日收盘 100，3 月 10 日除息，每股分红 2
    idx = pd.to_datetime(["2026-03-09", "2026-03-10"])
    close = pd.Series([100.0, 99.0], index=idx)
    f = D.dividend_factor(close, pd.Series([2.0], index=[pd.Timestamp("2026-03-10")]))
    assert f.tolist() == pytest.approx([0.98, 1.0])
    # 调整后的涨跌幅 = 99 / (100 × 0.98) − 1
    assert 99.0 / (100.0 * f.iloc[0]) - 1 == pytest.approx(0.010204, abs=1e-6)


def test_dividend_outside_data_range_is_ignored():
    close = daily([100.0, 101.0])["close"]
    later = pd.Series([1.0], index=[close.index[-1] + pd.Timedelta(days=30)])
    assert (D.dividend_factor(close, later) == 1.0).all()


# ---------- 体检 ----------

def test_health_check_on_clean_data_finds_nothing():
    report = D.health_check(make_bars(10), freq="1min", wick_window=4)
    assert D.summarize(report).sum() == 0


def test_health_check_finds_planted_problems():
    df = make_bars(10)
    df.iloc[2, df.columns.get_loc("high")] = 99.5                        # 最高价低于开盘价
    df.iloc[3, df.columns.get_loc("volume")] = 0.0                        # 零成交量
    df.iloc[4, df.columns.get_loc("close")] = np.nan                      # 空值
    df = df.drop(df.index[6])                                             # 缺一根
    shifted = df.index.tolist()
    shifted[-1] = shifted[-1] + pd.Timedelta(seconds=20)                  # 最后一根时间戳偏移
    df.index = pd.DatetimeIndex(shifted)
    df = pd.concat([df, df.iloc[[0]]]).sort_index()                       # 重复一根

    s = D.summarize(D.health_check(df, freq="1min", wick_window=4))
    assert s["高低价自相矛盾"] == 1
    assert s["成交量为零"] == 1
    assert s["有空值的行"] == 1
    assert s["缺失的 K 线"] == 1
    assert s["时间戳没有对齐"] == 1
    assert s["重复时间戳"] == 1


def test_missing_runs_groups_consecutive_gaps():
    idx = pd.DatetimeIndex(["2024-01-01 00:03", "2024-01-01 00:04", "2024-01-01 00:05", "2024-01-01 00:09"], tz="UTC")
    runs = D.missing_runs(idx, "1min")
    assert runs["缺失根数"].tolist() == [3, 1]


def test_fingerprint():
    fp = D.fingerprint(make_bars(3))
    assert fp["行数"] == 3 and fp["收盘价之和"] == 301.5

pytest -q

...............                                                          [100%]
15 passed in 0.29s

def price_to_y(price: float, lo: float, hi: float, top: float, height: float, log: bool = False) -> float:
    """把价格换算成图上的纵坐标（像素）。屏幕坐标向下增大，所以最高价在最上面。

    算术坐标：按价格的「差」等比例分配高度
    对数坐标：按价格的「比」等比例分配高度
    """
    if log:
        frac = (math.log(hi) - math.log(price)) / (math.log(hi) - math.log(lo))
    else:
        frac = (hi - price) / (hi - lo)
    return top + frac * height

btc = D.load_binance_klines(glob.glob("data/binance/spot/BTCUSDT/1d/*.zip"))
march = btc.loc["2020-03-01":"2020-03-31"]
lo, hi = march["low"].min(), march["high"].max()
print("最低价", lo, "最高价", hi, "K 线根数", len(march))
bar = march.loc["2020-03-13"]
for name in ["high", "close", "open", "low"]:
    y_lin = P.price_to_y(bar[name], lo, hi, top=20, height=360)
    y_log = P.price_to_y(bar[name], lo, hi, top=20, height=360, log=True)
    print(f"{name:>5} {bar[name]:>9.2f}  算术坐标 y = {y_lin:6.2f}  对数坐标 y = {y_log:6.2f}")

最低价 3782.13 最高价 9188.0 K 线根数 31
 high   5955.00  算术坐标 y = 235.30  对数坐标 y = 195.89
close   5578.60  算术坐标 y = 260.37  对数坐标 y = 222.37
 open   4800.01  算术坐标 y = 312.22  对数坐标 y = 283.34
  low   3782.13  算术坐标 y = 380.00  对数坐标 y = 380.00

<line x1="326.45" y1="235.30" x2="326.45" y2="380.00" stroke="#26a69a" stroke-width="1"/>
<rect x="317.87" y="260.37" width="17.16" height="51.85" fill="#26a69a"/>

# 涨跌颜色。国际惯例：涨绿跌红；中国大陆惯例：涨红跌绿
STYLES = {
    "international": {"up": "#26a69a", "down": "#ef5350"},
    "china": {"up": "#ef5350", "down": "#26a69a"},
}


def candles_svg(df: pd.DataFrame, width: int = 800, height: int = 400, log: bool = False,
                style: str = "international", pad: int = 20) -> str:
    """不用任何绘图库，直接生成一张 K 线图的 SVG 文本。"""
    colors = STYLES[style]
    lo, hi = df["low"].min(), df["high"].max()
    n = len(df)
    slot = (width - 2 * pad) / n                 # 每根 K 线占的水平宽度
    body_w = slot * 0.7                          # 实体宽度，留 30% 做间隔
    y = lambda p: price_to_y(p, lo, hi, pad, height - 2 * pad, log)

    parts = [f'<svg xmlns="http://www.w3.org/2000/svg" width="{width}" height="{height}">',
             f'<rect width="{width}" height="{height}" fill="white"/>']
    for i, (o, h, l, c) in enumerate(df[["open", "high", "low", "close"]].itertuples(index=False)):
        color = colors["up"] if c >= o else colors["down"]
        x_mid = pad + slot * (i + 0.5)
        # 影线：从最高价画到最低价的一条竖线
        parts.append(f'<line x1="{x_mid:.2f}" y1="{y(h):.2f}" x2="{x_mid:.2f}" y2="{y(l):.2f}" stroke="{color}" stroke-width="1"/>')
        # 实体：开盘价和收盘价之间的矩形；开盘等于收盘时至少画 1 像素高
        top, bottom = y(max(o, c)), y(min(o, c))
        parts.append(f'<rect x="{x_mid - body_w / 2:.2f}" y="{top:.2f}" width="{body_w:.2f}" '
                     f'height="{max(bottom - top, 1):.2f}" fill="{color}"/>')
    parts.append("</svg>")
    return "\n".join(parts)

Path("figures").mkdir(exist_ok=True)
svg = P.candles_svg(march)
Path("figures/hand-drawn.svg").write_text(svg)
lines = svg.splitlines()
print(len(lines), "行")
print(lines[26])          # 3 月 13 日的影线
print(lines[27])          # 3 月 13 日的实体

65 行
<line x1="326.45" y1="235.30" x2="326.45" y2="380.00" stroke="#26a69a" stroke-width="1"/>
<rect x="317.87" y="260.37" width="17.16" height="51.85" fill="#26a69a"/>

def plot_candles(df: pd.DataFrame, *, title: str | None = None, volume: bool = True, log: bool = False,
                 time_axis: str = "bars", style: str = "international",
                 overlays: dict[str, pd.Series] | None = None,
                 hlines: dict[str, float] | None = None,
                 marks: list[tuple] | None = None,
                 figsize=(10, 6), dpi: int = 150):
    """画 K 线图，返回 (fig, 价格子图, 成交量子图或 None)。

    time_axis: "bars" 每根 K 线等距排列（休市时段不留空白）；"real" 按真实时间排列
    overlays:  叠加在价格上的曲线，比如均线 {"MA20": series}
    hlines:    水平线 {"说明": 价格}
    marks:     标注点 [(时间, 价格, "文字"), ...]
    """
    colors = STYLES[style]
    if volume:
        fig, (ax, av) = plt.subplots(2, 1, figsize=figsize, dpi=dpi, sharex=True,
                                     gridspec_kw={"height_ratios": [3, 1]})
    else:
        fig, ax = plt.subplots(figsize=figsize, dpi=dpi)
        av = None

    if time_axis == "bars":
        xs = list(range(len(df)))
        width = 0.7
    elif time_axis == "real":
        xs = list(df.index)
        step = pd.Series(df.index).diff().median()      # 相邻两根 K 线通常相隔多久
        width = step * 0.7
    else:
        raise ValueError('time_axis 只能是 "bars" 或 "real"')
    x_of = dict(zip(df.index, xs))

    for x, (o, h, l, c, v) in zip(xs, df[["open", "high", "low", "close", "volume"]].itertuples(index=False)):
        color = colors["up"] if c >= o else colors["down"]
        ax.vlines(x, l, h, color=color, linewidth=0.8)
        ax.add_patch(Rectangle((x - width / 2, min(o, c)),
                               width, abs(c - o), facecolor=color, edgecolor=color, linewidth=0.5))
        if av is not None:
            av.bar(x, v, width=width, color=color)

    for name, series in (overlays or {}).items():
        s = series.reindex(df.index)
        ax.plot(xs, s.values, linewidth=1.2, label=name)
    for name, price in (hlines or {}).items():
        ax.axhline(price, linestyle=":", linewidth=1.2, color="#1f77b4")
        ax.annotate(name, (xs[0], price), textcoords="offset points", xytext=(2, 3), fontsize=9, color="#1f77b4")
    for t, price, text in (marks or []):
        # 文字放在点的下方，箭头向上指着这个价格
        ax.annotate(text, (x_of[pd.Timestamp(t, tz=df.index.tz) if df.index.tz else pd.Timestamp(t)], price),
                    textcoords="offset points", xytext=(0, -32), ha="center", fontsize=9,
                    arrowprops={"arrowstyle": "->", "color": "#555"})

    ax.autoscale_view()
    if log:
        ax.set_yscale("log")
        lo, hi = ax.get_ylim()
        # 价格跨度超过 20 倍时，刻度只放在 1、2、5 的倍数上；跨度小时放在 1～9 的倍数上，否则刻度太少
        subs = (1.0, 2.0, 5.0) if hi / lo > 20 else tuple(range(1, 10))
        ax.yaxis.set_major_locator(LogLocator(base=10, subs=subs))
        ax.yaxis.set_minor_locator(LogLocator(base=10, subs=()))
    # 刻度显示成带千位分隔的普通数字，不用 10 的几次方
    ax.yaxis.set_major_formatter(FuncFormatter(lambda v, _: f"{v:,.0f}" if v >= 100 else f"{v:g}"))
    if time_axis == "bars":
        ticks = _date_ticks(df.index)
        (av or ax).set_xticks(ticks)
        (av or ax).set_xticklabels([df.index[i].strftime("%Y-%m-%d") for i in ticks], fontsize=8)
        ax.set_xlim(-1, len(df))
    else:
        locator = mdates.AutoDateLocator(maxticks=8)
        (av or ax).xaxis.set_major_locator(locator)
        (av or ax).xaxis.set_major_formatter(mdates.ConciseDateFormatter(locator))
    if overlays:
        ax.legend(fontsize=9, loc="upper left")
    if title:
        ax.set_title(title, fontsize=12)
    ax.set_ylabel("价格")
    if av is not None:
        av.set_ylabel("成交量")
        av.yaxis.set_major_formatter(FuncFormatter(lambda v, _: f"{v:,.0f}"))
    for a in (ax, av):
        if a is not None:
            a.grid(alpha=0.25)
    fig.tight_layout()
    return fig, ax, av


def _date_ticks(index: pd.DatetimeIndex, n: int = 6) -> list[int]:
    """在等距排列的 K 线里，挑出大约 n 个位置放日期标签。"""
    step = max(1, len(index) // n)
    return list(range(0, len(index), step))


def use_chinese_font(path: str = "/System/Library/Fonts/Hiragino Sans GB.ttc") -> None:
    """让 matplotlib 能显示中文。默认路径是 macOS 自带字体；Windows 可以用 C:/Windows/Fonts/msyh.ttc。"""
    from matplotlib import font_manager
    font_manager.fontManager.addfont(path)
    plt.rcParams["font.family"] = font_manager.FontProperties(fname=path).get_name()
    plt.rcParams["axes.unicode_minus"] = False

"""talab.plot：K 线绘图。第 4 篇。"""
from __future__ import annotations

import math

import matplotlib.dates as mdates
import matplotlib.pyplot as plt
import pandas as pd
from matplotlib.patches import Rectangle
from matplotlib.ticker import FuncFormatter, LogLocator

P.use_chinese_font()
fig, ax, av = P.plot_candles(march, title="BTCUSDT 日线（2020 年 3 月），用 talab.plot 画")
fig.savefig("figures/talab-plot.png")

pip install mplfinance==0.12.10b0

import mplfinance as mpf
mpf.plot(march, type="candle", volume=True, style="yahoo", savefig="figures/mplfinance.png")

"""talab.plot 的测试。"""
import math

import matplotlib
import pandas as pd
import pytest

matplotlib.use("Agg")                 # 不弹出窗口

from talab import plot as P


def test_price_to_y_linear_endpoints_and_middle():
    # 价格区间 100～200，画在从像素 20 开始、高 400 的区域里
    assert P.price_to_y(200, 100, 200, 20, 400) == 20            # 最高价在最上面
    assert P.price_to_y(100, 100, 200, 20, 400) == 420           # 最低价在最下面
    assert P.price_to_y(150, 100, 200, 20, 400) == 220           # 算术中点在正中间


def test_price_to_y_log_puts_geometric_mean_in_the_middle():
    mid = math.sqrt(100 * 200)                                    # 141.42
    assert P.price_to_y(mid, 100, 200, 20, 400, log=True) == pytest.approx(220)


def test_log_scale_gives_equal_ratios_equal_distances():
    d1 = P.price_to_y(10, 10, 1000, 0, 300, log=True) - P.price_to_y(20, 10, 1000, 0, 300, log=True)
    d2 = P.price_to_y(100, 10, 1000, 0, 300, log=True) - P.price_to_y(200, 10, 1000, 0, 300, log=True)
    assert d1 == pytest.approx(d2)


def bars():
    idx = pd.date_range("2024-01-01", periods=3, freq="1D", tz="UTC")
    return pd.DataFrame({"open": [10, 12, 11], "high": [13, 13, 12], "low": [9, 10, 10],
                         "close": [12, 11, 11], "volume": [5, 6, 7]}, index=idx, dtype=float)


def test_candles_svg_draws_one_line_and_one_body_per_bar():
    svg = P.candles_svg(bars())
    assert svg.count("<line") == 3
    assert svg.count("<rect") == 3 + 1                            # 3 个实体 + 1 个白色背景


def test_candles_svg_colors_follow_style():
    svg = P.candles_svg(bars(), style="china")
    # 第一根收盘高于开盘，第三根收盘等于开盘（也按阳线画），每根的影线和实体各用一次颜色
    assert svg.count(P.STYLES["china"]["up"]) == 4
    assert svg.count(P.STYLES["china"]["down"]) == 2


def test_plot_candles_log_axis_and_volume():
    fig, ax, av = P.plot_candles(bars(), log=True)
    assert ax.get_yscale() == "log"
    assert len(ax.patches) == 3 and len(av.patches) == 3

for a, b in [(10, 20), (100, 200), (1000, 2000)]:
    d_lin = P.price_to_y(a, 10, 2000, 0, 300) - P.price_to_y(b, 10, 2000, 0, 300)
    d_log = P.price_to_y(a, 10, 2000, 0, 300, log=True) - P.price_to_y(b, 10, 2000, 0, 300, log=True)
    print(f"{a:>5} → {b:<5} 算术坐标上相距 {d_lin:6.2f} 像素，对数坐标上相距 {d_log:6.2f} 像素")

   10 → 20    算术坐标上相距   1.51 像素，对数坐标上相距  39.25 像素
  100 → 200   算术坐标上相距  15.08 像素，对数坐标上相距  39.25 像素
 1000 → 2000  算术坐标上相距 150.75 像素，对数坐标上相距  39.25 像素

x = btc.loc["2025-09-01":"2026-05-15"]
t1, p1 = pd.Timestamp("2025-10-06", tz="UTC"), btc.loc["2025-10-06", "high"]
t2, p2 = pd.Timestamp("2026-01-14", tz="UTC"), btc.loc["2026-01-14", "high"]
a = (x.index - t1) / (t2 - t1)                   # 第一个高点记为 0，第二个高点记为 1
linear = pd.Series(p1 + (p2 - p1) * a, index=x.index)
logline = pd.Series(np.exp(np.log(p1) + (np.log(p2) - np.log(p1)) * a), index=x.index)
after = x.index > t2
first_lin = x.index[after & (x["close"] > linear)][0]
first_log = x.index[after & (x["close"] > logline)][0]
print("两个高点：", p1, p2)
for day in [first_lin, first_log]:
    print(day.date(), "收盘", x.loc[day, "close"], " 算术直线", round(linear[day], 2), " 对数直线", round(logline[day], 2))
zero_day = t1 + (t2 - t1) * (p1 / (p1 - p2))
print("算术直线降到 0 的日期：", zero_day.date())

两个高点： 126199.63 97924.49
2026-04-13 收盘 74417.99  算术直线 72759.62  对数直线 78134.51
2026-04-22 收盘 78178.23  算术直线 70214.85  对数直线 76370.9
算术直线降到 0 的日期： 2026-12-26