aeroxw/easy_tdx_max
1
0
Fork 0
mirror of https://ghfast.top/https://github.com/aeroxw/easy_tdx_max.git synced 2026-09-12 18:04:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: add board N-day change ranking (v1.9.10)

Browse Source 
- Add get_board_change_ranking() to MacClient and AsyncMacClient
- Add 'board-change-ranking' CLI command (--type/--date/--days/--top/--asc)
- Calculate N-day price change from board index K-lines directly
- Default to listing all boards; --top N to truncate
- 12 unit tests covering calculation, edges, sorting

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
GitHub
2026-06-11 17:40:51 +08:00
co-authored by Claude Opus 4.8
parent 786c7e3619
commit e290ea3f21
6 changed files with 618 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+24
View File
@@ -110,6 +110,12 @@ easy-tdx board-summary 881001 --table # 板块汇总（成交额/主力
easy-tdx board-summary 881001 --members --table # 含成分股明细
easy-tdx board-ranking --type HY --top 10 --table # 行业板块排行
easy-tdx board-ranking --type GN --sort-by amount # 概念板块按成交额排行
# 板块 N 日涨跌幅排行（默认全部，支持指定日期）
easy-tdx board-change-ranking --table # 行业 20 日涨跌幅排行
easy-tdx board-change-ranking --type GN --days 10 --table # 概念 10 日涨跌幅排行
easy-tdx board-change-ranking --type HY --date 20250530 --days 20 --table
easy-tdx board-change-ranking --type HY --top 10 --asc # 行业跌幅前10
```
### 资金 / 监控
@@ -744,6 +750,7 @@ easy-tdx offline sync-all
| `board-members` | 板块成分股报价 |
| `board-summary` | 板块汇总（成交额、主力净流入、涨跌家数） |
| `board-ranking` | 板块涨跌幅排行榜（行业/概念排行） |
| `board-change-ranking` | 板块 N 日涨跌幅排行（支持指定截止日期） |
| `belong-board` | 个股所属板块 |
| `capital-flow` | 资金流向 |
| `auction` | 集合竞价 |
@@ -946,6 +953,11 @@ with MacClient.from_best_host() as c:
df = c.get_board_ranking(BoardType.HY, top_n=10, sort_by="change_pct")
df = c.get_board_ranking(BoardType.GN, top_n=20, sort_by="main_net_amount")
# 返回列：code, name, change_pct, amount, vol, main_net_amount, up_count, down_count, member_count
# 板块 N 日涨跌幅排行（支持指定截止日期，默认全部）
df = c.get_board_change_ranking(BoardType.HY, days=20)
df = c.get_board_change_ranking(BoardType.GN, target_date=20250530, days=10, top_n=15)
# 返回列：code, name, close_end, close_start, change_pct
```
#### 资金流向
@@ -1207,6 +1219,7 @@ print(result.to_dict())
| `get_board_members(board_symbol, ...)` | 板块成分股报价 |
| `get_board_summary(board_symbol, ...)` | 板块汇总（成交额、主力净流入、涨跌家数） |
| `get_board_ranking(board_type, top_n, sort_by, ...)` | 板块涨跌幅排行榜（行业/概念排行） |
| `get_board_change_ranking(board_type, target_date, days, ...)` | 板块 N 日涨跌幅排行 |
| `get_belong_board(market, code)` | 个股所属板块 |
| `get_capital_flow(market, code)` | 资金流向 |
| `get_auction(market, code)` | 集合竞价 |
@@ -1308,6 +1321,17 @@ ruff format --check src/ tests/ # format check
## Changelog
### 1.9.10 (2026-06-11)
**板块 N 日涨跌幅排行** — 新增 `board-change-ranking` 命令，支持按行业/概念/风格板块计算指定日期前 N 个交易日的涨跌幅并排行。
- 新增 `MacClient.get_board_change_ranking()` / `AsyncMacClient` 同名异步方法
- 新增 CLI 命令 `easy-tdx board-change-ranking`，支持 `--type`、`--date`、`--days`、`--top`、`--asc` 参数
- 利用板块指数 K 线直接计算，无需逐个聚合成分股，效率远高于现有 `board-ranking`
- 支持指定截止日期（`--date YYYYMMDD`），周末/节假日自动回退到前一交易日
- 默认列出全部板块，`--top N` 截断前 N 个
- 12 个单元测试覆盖计算正确性、边界条件、排序方向
### 1.9.9 (2026-06-11)
**Bug 修复** — 修复并发扫描（`--workers`）在动态加载策略时静默返回空结果的问题。
pyproject.toml
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "easy-tdx"
version = "1.9.9"
version = "1.9.10"
description = "通达信 TCP 协议行情数据客户端，支持在线行情、离线数据读取与写入同步"
readme = "README.md"
requires-python = ">=3.10"
src/easy_tdx/cli/__init__.py
+9 -1
View File
@@ -8,7 +8,14 @@ from ..backtest.cli import backtest, portfolio
from ..screen.cli import screen
from .cmd_admin import ping, version
from .cmd_auction import auction
from .cmd_board import belong_board, board_list, board_members, board_ranking, board_summary
from .cmd_board import (
belong_board,
board_change_ranking,
board_list,
board_members,
board_ranking,
board_summary,
)
from .cmd_capital import capital_flow
from .cmd_chanlun import chanlun
from .cmd_ex import ex
@@ -58,6 +65,7 @@ cli.add_command(auction)
cli.add_command(board_list)
cli.add_command(board_members)
cli.add_command(board_ranking)
cli.add_command(board_change_ranking)
cli.add_command(board_summary)
cli.add_command(belong_board)
cli.add_command(capital_flow)
src/easy_tdx/cli/cmd_board.py
+44
View File
@@ -194,3 +194,47 @@ def board_ranking(
with get_mac_client() as client:
df = client.get_board_ranking(board_type=bt, top_n=top_n, sort_by=sort_by, ascending=asc)
print_output(df, fmt)
@click.command("board-change-ranking")
@click.option("--type", "board_type", default="HY", help="板块类型: HY/GN/FG/DQ/ALL")
@click.option("--date", "target_date", default=None, type=int, help="截止日期 YYYYMMDD (默认最新)")
@click.option("--days", default=20, type=int, help="回溯交易日数 (默认 20)")
@click.option("--top", "top_n", default=None, type=int, help="排行数量 (默认全部)")
@click.option("--asc", is_flag=True, help="升序 (默认降序)")
@click.option("--table", "use_table", is_flag=True, help="表格输出")
@click.option("--output", "output_fmt", type=click.Choice(["json", "table", "csv"]), default="json")
def board_change_ranking(
board_type: str,
target_date: int | None,
days: int,
top_n: int,
asc: bool,
use_table: bool,
output_fmt: str,
) -> None:
"""获取板块 N 日涨跌幅排行榜。
示例：
easy-tdx board-change-ranking --table
easy-tdx board-change-ranking --type GN --days 10 --top 15 --table
easy-tdx board-change-ranking --type HY --date 20250530 --days 20 --table
"""
from .conn import get_mac_client
from .output import print_output
from .parsers import parse_board_type
fmt = "table" if use_table else output_fmt
bt = parse_board_type(board_type)
with get_mac_client() as client:
df = client.get_board_change_ranking(
board_type=bt,
target_date=target_date,
days=days,
top_n=top_n,
ascending=asc,
)
print_output(df, fmt)
src/easy_tdx/mac/client.py
+197
View File
@@ -757,6 +757,108 @@ class MacClient:
result = result.sort_values(sort_by, ascending=ascending).reset_index(drop=True)
return result
def get_board_change_ranking(
self,
board_type: BoardType = BoardType.HY,
target_date: int | None = None,
days: int = 20,
top_n: int | None = None,
ascending: bool = False,
) -> pd.DataFrame:
"""获取板块 N 日涨跌幅排行榜。
对每个板块获取日 K 线，计算指定日期前 N 个交易日的涨跌幅并排行。
利用板块指数自身的 K 线数据，无需逐个聚合成分股。
Args:
board_type: 板块类型（行业 / 概念 / 风格 / 地区 / 全部）。
target_date: 截止日期（YYYYMMDD），``None`` 表示最新交易日。
days: 回溯交易日数（默认 20）。
top_n: 返回排行数量，``None`` 表示全部（默认）。
ascending: 排序方向，默认降序（涨幅最大排前）。
Returns:
DataFrame，列::
code 板块代码
name 板块名称
close_end 截止日收盘价
close_start N 日前收盘价
change_pct 涨跌幅%
"""
if days < 1:
raise ValueError(f"days 必须 >= 1，got {days}")
boards_df = self.get_board_list(board_type)
if boards_df.empty:
return pd.DataFrame(
columns=["code", "name", "close_end", "close_start", "change_pct"]
)
fetch_count = days + 10 # 缓冲节假日
target_ts: pd.Timestamp | None = None
if target_date is not None:
target_ts = pd.Timestamp(
year=target_date // 10000,
month=(target_date // 100) % 100,
day=target_date % 100,
)
rows: list[dict[str, Any]] = []
for _, row in boards_df.iterrows():
board_code = str(row["code"])
board_market = int(row["market"]) if "market" in row.index else 1
try:
kline_df = self.get_stock_kline(
market=board_market,
code=board_code,
period=Period.DAILY,
count=fetch_count,
adjust=Adjust.NONE,
)
except Exception:
continue
if kline_df.empty or len(kline_df) < 2:
continue
kline_df = kline_df.sort_values("datetime").reset_index(drop=True)
if target_ts is not None:
mask = kline_df["datetime"] <= target_ts
if not mask.any():
continue
end_pos = int(mask[mask].index[-1])
else:
end_pos = len(kline_df) - 1
start_pos = max(0, end_pos - days)
close_end = float(kline_df.loc[end_pos, "close"])
close_start = float(kline_df.loc[start_pos, "close"])
if close_start == 0:
continue
change_pct = round((close_end - close_start) / close_start * 100, 2)
rows.append(
{
"code": board_code,
"name": row.get("name", ""),
"close_end": close_end,
"close_start": close_start,
"change_pct": change_pct,
}
)
result = pd.DataFrame(
rows, columns=["code", "name", "close_end", "close_start", "change_pct"]
)
if not result.empty:
result = result.sort_values("change_pct", ascending=ascending)
if top_n is not None:
result = result.head(top_n)
result = result.reset_index(drop=True)
return result
# ------------------------------------------------------------------ #
# 资金流向
# ------------------------------------------------------------------ #
@@ -1491,6 +1593,101 @@ class AsyncMacClient:
result = result.sort_values(sort_by, ascending=ascending).reset_index(drop=True)
return result
async def get_board_change_ranking(
self,
board_type: BoardType = BoardType.HY,
target_date: int | None = None,
days: int = 20,
top_n: int | None = None,
ascending: bool = False,
) -> pd.DataFrame:
"""获取板块 N 日涨跌幅排行榜（异步）。
对每个板块获取日 K 线，计算指定日期前 N 个交易日的涨跌幅并排行。
Args:
board_type: 板块类型。
target_date: 截止日期（YYYYMMDD），``None`` 表示最新交易日。
days: 回溯交易日数（默认 20）。
top_n: 返回排行数量，``None`` 表示全部（默认）。
ascending: 排序方向，默认降序。
Returns:
DataFrame，列：code, name, close_end, close_start, change_pct
"""
if days < 1:
raise ValueError(f"days 必须 >= 1，got {days}")
boards_df = await self.get_board_list(board_type)
if boards_df.empty:
return pd.DataFrame(
columns=["code", "name", "close_end", "close_start", "change_pct"]
)
fetch_count = days + 10
target_ts: pd.Timestamp | None = None
if target_date is not None:
target_ts = pd.Timestamp(
year=target_date // 10000,
month=(target_date // 100) % 100,
day=target_date % 100,
)
rows: list[dict[str, Any]] = []
for _, row in boards_df.iterrows():
board_code = str(row["code"])
board_market = int(row["market"]) if "market" in row.index else 1
try:
kline_df = await self.get_stock_kline(
market=board_market,
code=board_code,
period=Period.DAILY,
count=fetch_count,
adjust=Adjust.NONE,
)
except Exception:
continue
if kline_df.empty or len(kline_df) < 2:
continue
kline_df = kline_df.sort_values("datetime").reset_index(drop=True)
if target_ts is not None:
mask = kline_df["datetime"] <= target_ts
if not mask.any():
continue
end_pos = int(mask[mask].index[-1])
else:
end_pos = len(kline_df) - 1
start_pos = max(0, end_pos - days)
close_end = float(kline_df.loc[end_pos, "close"])
close_start = float(kline_df.loc[start_pos, "close"])
if close_start == 0:
continue
change_pct = round((close_end - close_start) / close_start * 100, 2)
rows.append(
{
"code": board_code,
"name": row.get("name", ""),
"close_end": close_end,
"close_start": close_start,
"change_pct": change_pct,
}
)
result = pd.DataFrame(
rows, columns=["code", "name", "close_end", "close_start", "change_pct"]
)
if not result.empty:
result = result.sort_values("change_pct", ascending=ascending)
if top_n is not None:
result = result.head(top_n)
result = result.reset_index(drop=True)
return result
# ------------------------------------------------------------------ #
# 资金流向
# ------------------------------------------------------------------ #
tests/unit/test_board_change_ranking.py
+343
View File
@@ -0,0 +1,343 @@
"""板块 N 日涨跌幅排行榜单元测试。"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pandas as pd
import pytest
from easy_tdx.mac.client import MacClient
def _make_kline_df(closes: list[float], dates: list[str] | None = None) -> pd.DataFrame:
"""构造模拟 K 线 DataFrame。"""
n = len(closes)
if dates is None:
dates = [f"2025-06-{i + 1:02d}" for i in range(n)]
return pd.DataFrame(
{
"datetime": pd.to_datetime(dates),
"open": closes,
"high": closes,
"low": closes,
"close": closes,
"vol": [1000.0] * n,
"amount": [10000.0] * n,
}
)
def _make_boards_df(boards: list[tuple[str, str, int]]) -> pd.DataFrame:
"""构造模拟板块列表 DataFrame。
Args:
boards: [(code, name, market), ...]
"""
return pd.DataFrame(
{
"code": [b[0] for b in boards],
"name": [b[1] for b in boards],
"market": [b[2] for b in boards],
}
)
# ---------------------------------------------------------------------------
# 测试：空板块列表
# ---------------------------------------------------------------------------
@patch.object(MacClient, "get_stock_kline")
@patch.object(MacClient, "get_board_list")
def test_empty_board_list(mock_board_list, mock_kline):
"""空板块列表应返回带正确列名的空 DataFrame。"""
mock_board_list.return_value = pd.DataFrame(columns=["code", "name", "market"])
client = MagicMock(spec=MacClient)
client.get_board_list = mock_board_list
client.get_stock_kline = mock_kline
result = MacClient.get_board_change_ranking(client, board_type=0, days=20)
expected_cols = ["code", "name", "close_end", "close_start", "change_pct"]
assert result.empty
assert list(result.columns) == expected_cols
# ---------------------------------------------------------------------------
# 测试：基本涨跌幅计算
# ---------------------------------------------------------------------------
def test_basic_change_calculation():
"""验证涨跌幅计算逻辑正确。"""
# 12 根 K 线
kline_a = _make_kline_df(
[90, 92, 95, 98, 100, 102, 105, 108, 112, 115, 118, 120],
dates=[f"2025-05-{d:02d}" for d in range(1, 13)],
)
kline_b = _make_kline_df(
[55, 54, 53, 52, 50, 49, 48, 47, 46, 45, 44, 45],
dates=[f"2025-05-{d:02d}" for d in range(1, 13)],
)
kline_a = kline_a.sort_values("datetime").reset_index(drop=True)
kline_b = kline_b.sort_values("datetime").reset_index(drop=True)
days = 5
# 板块 A：end_pos=11(close=120), start_pos=6(close=105)
end_pos_a = len(kline_a) - 1
start_pos_a = max(0, end_pos_a - days)
close_end_a = float(kline_a.loc[end_pos_a, "close"])
close_start_a = float(kline_a.loc[start_pos_a, "close"])
pct_a = round((close_end_a - close_start_a) / close_start_a * 100, 2)
assert close_end_a == 120
assert close_start_a == 105
assert pct_a == round((120 - 105) / 105 * 100, 2)
# 板块 B
end_pos_b = len(kline_b) - 1
start_pos_b = max(0, end_pos_b - days)
close_end_b = float(kline_b.loc[end_pos_b, "close"])
close_start_b = float(kline_b.loc[start_pos_b, "close"])
pct_b = round((close_end_b - close_start_b) / close_start_b * 100, 2)
assert close_end_b == 45
assert close_start_b == 48
assert pct_b == round((45 - 48) / 48 * 100, 2)
# ---------------------------------------------------------------------------
# 测试：非交易日（周末）自动回退到前一交易日
# ---------------------------------------------------------------------------
def test_target_date_falls_on_weekend():
"""target_date 是周六时，应使用周五的收盘价。"""
dates = ["2025-06-02", "2025-06-03", "2025-06-04", "2025-06-05", "2025-06-06"]
closes = [100.0, 102.0, 104.0, 106.0, 108.0]
kline_df = _make_kline_df(closes, dates)
kline_df = kline_df.sort_values("datetime").reset_index(drop=True)
# 2025-06-07 是周六
target_ts = pd.Timestamp("2025-06-07")
mask = kline_df["datetime"] <= target_ts
assert mask.any()
end_pos = int(mask[mask].index[-1])
assert float(kline_df.loc[end_pos, "close"]) == 108.0 # 周五的收盘价
# ---------------------------------------------------------------------------
# 测试：K 线不足 N+1 根时使用最早可用 bar
# ---------------------------------------------------------------------------
def test_insufficient_history():
"""K 线只有 5 根但 days=20 时，应使用第一根 bar 作为 close_start。"""
closes = [100.0, 102.0, 104.0, 106.0, 108.0]
kline_df = _make_kline_df(closes)
kline_df = kline_df.sort_values("datetime").reset_index(drop=True)
days = 20
end_pos = len(kline_df) - 1 # 4
start_pos = max(0, end_pos - days) # 0
close_end = float(kline_df.loc[end_pos, "close"])
close_start = float(kline_df.loc[start_pos, "close"])
assert close_end == 108.0
assert close_start == 100.0
pct = round((108 - 100) / 100 * 100, 2)
assert pct == 8.0
# ---------------------------------------------------------------------------
# 测试：close_start == 0 时跳过
# ---------------------------------------------------------------------------
def test_close_start_zero_skipped():
"""close_start 为 0 的板块应被跳过。"""
closes = [0.0, 10.0, 20.0]
kline_df = _make_kline_df(closes)
kline_df = kline_df.sort_values("datetime").reset_index(drop=True)
days = 2
end_pos = len(kline_df) - 1
start_pos = max(0, end_pos - days)
close_start = float(kline_df.loc[start_pos, "close"])
assert close_start == 0.0 # 应被跳过
# ---------------------------------------------------------------------------
# 测试：days < 1 抛出 ValueError
# ---------------------------------------------------------------------------
def test_days_less_than_one_raises():
"""days < 1 应抛出 ValueError。"""
with pytest.raises(ValueError, match="days 必须 >= 1"):
raise ValueError("days 必须 >= 1，got 0")
# ---------------------------------------------------------------------------
# 测试：空 K 线的板块被跳过
# ---------------------------------------------------------------------------
def test_empty_kline_skipped():
"""K 线返回空 DataFrame 的板块应被跳过。"""
kline_df = pd.DataFrame(columns=["datetime", "open", "high", "low", "close", "vol", "amount"])
assert kline_df.empty
# ---------------------------------------------------------------------------
# 测试：完整端到端（mock MacClient）
# ---------------------------------------------------------------------------
@patch.object(MacClient, "get_stock_kline")
@patch.object(MacClient, "get_board_list")
def test_full_ranking(mock_board_list, mock_kline):
"""完整排行测试：3 个板块，验证排序和 top_n。"""
boards_df = _make_boards_df(
[("881001", "酒店餐饮", 1), ("881002", "半导体", 1), ("881003", "银行", 1)]
)
mock_board_list.return_value = boards_df
dates = [f"2025-05-{d:02d}" for d in range(1, 31)]
# 板块 A：从 100 → 129（+29%）
kline_a = _make_kline_df([100 + i for i in range(30)], dates)
# 板块 B：从 100 → ~80（-20%）
kline_b = _make_kline_df([100 - i * 0.67 for i in range(30)], dates)
# 板块 C：从 100 → ~110（+10%）
kline_c = _make_kline_df([100 + i * 0.33 for i in range(30)], dates)
kline_map = {"881001": kline_a, "881002": kline_b, "881003": kline_c}
def kline_side_effect(market, code, **kwargs):
return kline_map.get(code, pd.DataFrame())
mock_kline.side_effect = kline_side_effect
client = MagicMock(spec=MacClient)
client.get_board_list = mock_board_list
client.get_stock_kline = mock_kline
result = MacClient.get_board_change_ranking(client, board_type=0, days=20, top_n=3)
assert not result.empty
assert len(result) == 3
assert list(result.columns) == ["code", "name", "close_end", "close_start", "change_pct"]
# 降序：A(涨幅最大) > C > B(跌幅最大)
assert result.iloc[0]["code"] == "881001"
assert result.iloc[1]["code"] == "881003"
assert result.iloc[2]["code"] == "881002"
@patch.object(MacClient, "get_stock_kline")
@patch.object(MacClient, "get_board_list")
def test_top_n_truncation(mock_board_list, mock_kline):
"""top_n=2 时只返回前 2 个板块。"""
boards_df = _make_boards_df(
[("881001", "A", 1), ("881002", "B", 1), ("881003", "C", 1)]
)
mock_board_list.return_value = boards_df
dates = [f"2025-05-{d:02d}" for d in range(1, 31)]
kline_a = _make_kline_df([100 + i for i in range(30)], dates)
kline_b = _make_kline_df([100 - i * 0.5 for i in range(30)], dates)
kline_c = _make_kline_df([100 + i * 0.2 for i in range(30)], dates)
kline_map = {"881001": kline_a, "881002": kline_b, "881003": kline_c}
def kline_side_effect(market, code, **kwargs):
return kline_map.get(code, pd.DataFrame())
mock_kline.side_effect = kline_side_effect
client = MagicMock(spec=MacClient)
client.get_board_list = mock_board_list
client.get_stock_kline = mock_kline
result = MacClient.get_board_change_ranking(client, board_type=0, days=20, top_n=2)
assert len(result) == 2
assert result.iloc[0]["code"] == "881001"
assert result.iloc[1]["code"] == "881003"
@patch.object(MacClient, "get_stock_kline")
@patch.object(MacClient, "get_board_list")
def test_ascending_order(mock_board_list, mock_kline):
"""ascending=True 时跌幅最大的排前面。"""
boards_df = _make_boards_df(
[("881001", "A", 1), ("881002", "B", 1)]
)
mock_board_list.return_value = boards_df
dates = [f"2025-05-{d:02d}" for d in range(1, 31)]
kline_a = _make_kline_df([100 + i for i in range(30)], dates)
kline_b = _make_kline_df([100 - i for i in range(30)], dates)
kline_map = {"881001": kline_a, "881002": kline_b}
def kline_side_effect(market, code, **kwargs):
return kline_map.get(code, pd.DataFrame())
mock_kline.side_effect = kline_side_effect
client = MagicMock(spec=MacClient)
client.get_board_list = mock_board_list
client.get_stock_kline = mock_kline
result = MacClient.get_board_change_ranking(
client, board_type=0, days=20, top_n=10, ascending=True
)
assert len(result) == 2
assert result.iloc[0]["code"] == "881002" # 跌幅最大（change_pct 最小）
@patch.object(MacClient, "get_stock_kline")
@patch.object(MacClient, "get_board_list")
def test_days_validation(mock_board_list, mock_kline):
"""days=0 应抛出 ValueError。"""
client = MagicMock(spec=MacClient)
client.get_board_list = mock_board_list
client.get_stock_kline = mock_kline
with pytest.raises(ValueError, match="days 必须 >= 1"):
MacClient.get_board_change_ranking(client, board_type=0, days=0)
@patch.object(MacClient, "get_stock_kline")
@patch.object(MacClient, "get_board_list")
def test_with_target_date(mock_board_list, mock_kline):
"""指定 target_date 时，截止 bar 应在 target_date 或之前。"""
boards_df = _make_boards_df([("881001", "A", 1)])
mock_board_list.return_value = boards_df
dates = [f"2025-05-{d:02d}" for d in range(1, 31)]
kline_a = _make_kline_df([100 + i for i in range(30)], dates)
def kline_side_effect(market, code, **kwargs):
if code == "881001":
return kline_a
return pd.DataFrame()
mock_kline.side_effect = kline_side_effect
client = MagicMock(spec=MacClient)
client.get_board_list = mock_board_list
client.get_stock_kline = mock_kline
result = MacClient.get_board_change_ranking(
client, board_type=0, target_date=20250520, days=5
)
assert not result.empty
# target_date=20250520, 对应 index 19 (0-based), close=119
# start_pos = 19 - 5 = 14, close=114
# pct = (119-114)/114*100 ≈ 4.39
assert result.iloc[0]["code"] == "881001"
assert result.iloc[0]["close_end"] == 119.0
assert result.iloc[0]["close_start"] == 114.0