From 4cb48daabb6fcd21d418a390e2d0eb4dfc24f0a2 Mon Sep 17 00:00:00 2001 From: M Date: Wed, 15 Apr 2026 21:23:13 +0800 Subject: [PATCH] Add live fallbacks for history fund flow and price limits --- README.md | 6 +- scripts/verify_fixes_20260415.py | 17 +- src/xmtdx/client.py | 339 ++++++++++++++++++-------- src/xmtdx/codec/price_rules.py | 56 ++++- tests/unit/test_a_share_extensions.py | 89 +++++++ tests/unit/test_protocol_fixes.py | 16 ++ 6 files changed, 402 insertions(+), 121 deletions(-) diff --git a/README.md b/README.md index 3807aee..6fed714 100644 --- a/README.md +++ b/README.md @@ -99,6 +99,7 @@ client = AsyncTdxClient.from_best_host(ping_timeout=5.0) | `get_security_list_all()` | 沪深 A 股列表（自动挂载行业信息；BJ 暂未纳入） | | `get_market_stat()` | 全市场 A 股涨跌统计（家数、成交额） | | `get_security_quotes([(market, code), ...])` | 批量实时五档行情（最多 80 只/次） | +| `get_price_limits(market, code, name, pre_close)` | 计算当前涨跌停价（自动处理上市初期无涨跌幅限制） | | `get_security_bars(market, code, category, start, count=800)` | K 线（股票） | | `get_index_bars(market, code, category, start, count=800)` | K 线（指数） | | `get_minute_time_data(market, code)` | 今日分时（240 条） | @@ -106,7 +107,7 @@ client = AsyncTdxClient.from_best_host(ping_timeout=5.0) | `get_transaction_data(market, code, start, count=800)` | 当日逐笔成交（分页） | | `get_history_transaction_data(market, code, date, start, count=800)` | 历史逐笔成交 | | `get_fund_flow(market, code)` | 当日资金流向统计（超大/大/中/小单） | -| `get_history_fund_flow(market, code, start, count)` | 历史日线资金流向序列（Category 22，实验性） | +| `get_history_fund_flow(market, code, start, count)` | 历史日线资金流向序列（优先 Category 22，空回包时自动回退到历史逐笔重算） | | `get_xdxr_info(market, code)` | 除权除息历史 | | `get_finance_info(market, code)` | 最新财务数据 | | `get_company_info_category(market, code)` | 公司信息文件目录 | @@ -149,7 +150,8 @@ _raw ``` `limit_up` / `limit_down` 当前不再直接由协议字段映射，默认保留为 `None`； -建议通过 `xmtdx.codec.price_rules.compute_price_limits(...)` 按业务规则计算。 +建议通过 `client.get_price_limits(...)` 计算当前涨跌停价，或用 +`xmtdx.codec.price_rules.compute_price_limits(..., listed_days=...)` 做纯规则计算。 ### MinuteBar（分时） diff --git a/scripts/verify_fixes_20260415.py b/scripts/verify_fixes_20260415.py index 29a42c5..3f172a6 100644 --- a/scripts/verify_fixes_20260415.py +++ b/scripts/verify_fixes_20260415.py @@ -3,7 +3,6 @@ import sys from xmtdx import Market, TdxClient -from xmtdx.codec.price_rules import compute_price_limits from xmtdx.models.enums import KlineCategory @@ -58,8 +57,8 @@ def main(): print(f" Error: {e}") success = False - # 3. 验证价格规则引擎 - print("\n[3] Price Limits (Rule Engine):") + # 3. 验证价格限制计算 + print("\n[3] Price Limits:") samples = [ ("600000", Market.SH, "浦发银行"), ("300750", Market.SZ, "宁德时代"), @@ -69,7 +68,7 @@ def main(): try: quotes = client.get_security_quotes([(market, code) for code, market, _name in samples]) for q, (_code, _market, name) in zip(quotes, samples, strict=True): - lu, ld = compute_price_limits(q.market, q.code, name, q.pre_close) + lu, ld = client.get_price_limits(q.market, q.code, name, q.pre_close) print( f" {q.code}: Price={q.price:.2f}, PreClose={q.pre_close:.2f}, " f"LimitUp={lu}, LimitDown={ld}" @@ -85,8 +84,8 @@ def main(): print(f" Error: {e}") success = False - # 4. 验证 get_history_fund_flow (Category 22) - print("\n[4] History Fund Flow (Category 22, experimental):") + # 4. 验证 get_history_fund_flow（直连或 fallback） + print("\n[4] History Fund Flow:") try: h_flow = client.get_history_fund_flow(Market.SH, "600000", 0, 1) if h_flow: @@ -94,9 +93,11 @@ def main(): print(f" Date: {f.year}-{f.month}-{f.day}, SuperIn: {f.super_in:.2f}") print(" Result: SUCCESS") else: - print(" Result: INFO (No data returned; interface remains experimental)") + print(" Result: FAIL (No data returned)") + success = False except Exception as e: - print(f" Error: {e} (Experimental interface; not counted as hard failure)") + print(f" Error: {e}") + success = False # 5. 验证 get_fund_flow 分页 print("\n[5] Fund Flow Pagination (600000):") diff --git a/src/xmtdx/client.py b/src/xmtdx/client.py index a7d2a5b..402b797 100644 --- a/src/xmtdx/client.py +++ b/src/xmtdx/client.py @@ -1,11 +1,13 @@ """高层行情 API：TdxClient（同步）和 AsyncTdxClient（asyncio）。""" import asyncio +from collections.abc import Awaitable, Callable from types import TracebackType from typing import TypeVar from .codec.block import parse_block_dat from .codec.industry import parse_tdxhy_cfg +from .codec.price_rules import compute_price_limits, get_no_limit_window_days from .commands.base import BaseCommand from .commands.block_info import GetBlockInfoCmd, GetBlockInfoMetaCmd from .commands.company_info import GetCompanyInfoCategoryCmd, GetCompanyInfoContentCmd @@ -34,6 +36,83 @@ _DEFAULT_PORT = 7709 _T = TypeVar("_T") +def _record_signature( + record: TransactionRecord, +) -> tuple[int, int, float, int, int, int]: + return ( + record.hour, + record.minute, + record.price, + record.vol, + record.buyorsell, + record.unknown_last, + ) + + +def _page_signature( + records: list[TransactionRecord], +) -> tuple[tuple[int, int, float, int, int, int], tuple[int, int, float, int, int, int]]: + return (_record_signature(records[0]), _record_signature(records[-1])) + + +def _classify_fund_flow(records: list[TransactionRecord]) -> FundFlow: + stats = { + "super_in": 0.0, + "large_in": 0.0, + "medium_in": 0.0, + "small_in": 0.0, + "super_out": 0.0, + "large_out": 0.0, + "medium_out": 0.0, + "small_out": 0.0, + } + + for record in records: + amount = record.price * record.vol * 100.0 + direction = ( + "in" if record.buyorsell == 0 else "out" if record.buyorsell == 1 else None + ) + if not direction: + continue + + if amount >= 1_000_000: + stats[f"super_{direction}"] += amount + elif amount >= 200_000: + stats[f"large_{direction}"] += amount + elif amount >= 40_000: + stats[f"medium_{direction}"] += amount + else: + stats[f"small_{direction}"] += amount + + return FundFlow(**stats) + + +def _date_from_bar(bar: SecurityBar) -> int: + return bar.year * 10000 + bar.month * 100 + bar.day + + +def _historical_fund_flow_from_records( + date: int, records: list[TransactionRecord] +) -> HistoricalFundFlow: + flow = _classify_fund_flow(records) + year = date // 10000 + month = (date // 100) % 100 + day = date % 100 + return HistoricalFundFlow( + year=year, + month=month, + day=day, + super_in=flow.super_in, + super_out=flow.super_out, + large_in=flow.large_in, + large_out=flow.large_out, + medium_in=flow.medium_in, + medium_out=flow.medium_out, + small_in=flow.small_in, + small_out=flow.small_out, + ) + + # ============================================================ # 同步客户端 # ============================================================ @@ -195,6 +274,32 @@ class TdxClient: """批量获取实时五档行情（最多80只/次）。""" return self._execute(GetSecurityQuotesCmd(stocks)) + def get_price_limits( + self, market: Market, code: str, name: str, pre_close: float + ) -> tuple[float | None, float | None]: + """按当前交易状态计算涨跌停价。 + + 对上市初期不设涨跌幅限制的标的，会先用日 K 线条数估算已上市交易天数。 + """ + listed_days: int | None = None + no_limit_window_days = get_no_limit_window_days(market, code, name) + if no_limit_window_days > 0: + try: + bars = self.get_security_bars( + market, code, KlineCategory.DAY, 0, no_limit_window_days + 1 + ) + listed_days = len(bars) + except Exception: + listed_days = None + + return compute_price_limits( + market, + code, + name, + pre_close, + listed_days=listed_days, + ) + # ------------------------------------------------------------------ # # K 线 # ------------------------------------------------------------------ # @@ -340,81 +445,84 @@ class TdxClient: total_volume=q.vol, ) - def get_fund_flow(self, market: Market, code: str) -> FundFlow: - """获取个股当日资金流向分布（基于 L1 逐笔数据统计）。""" - # 1. 分页拉取当日分笔并去重 + def _collect_transaction_records( + self, + fetch_page: Callable[[int, int], list[TransactionRecord]], + page_size: int, + max_start: int = 10000, + ) -> list[TransactionRecord]: all_recs: list[TransactionRecord] = [] - seen_sig = set() - seen_page_sigs = set() + seen_sig: set[tuple[int, int, float, int, int, int]] = set() + seen_page_sigs: set[ + tuple[ + tuple[int, int, float, int, int, int], + tuple[int, int, float, int, int, int], + ] + ] = set() start = 0 - - while start < 10000: - recs = self.get_transaction_data(market, code, start, 2000) + + while start < max_start: + recs = fetch_page(start, page_size) if not recs: break - - # 页签名判断：首尾记录组合 - page_sig = ( - ( - recs[0].hour, recs[0].minute, recs[0].price, - recs[0].vol, recs[0].buyorsell, recs[0].unknown_last - ), - ( - recs[-1].hour, recs[-1].minute, recs[-1].price, - recs[-1].vol, recs[-1].buyorsell, recs[-1].unknown_last - ), - ) + + page_sig = _page_signature(recs) if page_sig in seen_page_sigs: break seen_page_sigs.add(page_sig) new_count = 0 - for r in recs: - sig = (r.hour, r.minute, r.price, r.vol, r.buyorsell, r.unknown_last) + for record in recs: + sig = _record_signature(record) if sig not in seen_sig: seen_sig.add(sig) - all_recs.append(r) + all_recs.append(record) new_count += 1 - + if new_count == 0: break - + start += len(recs) if len(recs) < 100: break - - # 2. 统计逻辑 - # A 股标准：超大(>100w), 大单(20w-100w), 中单(4w-20w), 小单(<4w) - stats = { - "super_in": 0.0, "large_in": 0.0, "medium_in": 0.0, "small_in": 0.0, - "super_out": 0.0, "large_out": 0.0, "medium_out": 0.0, "small_out": 0.0, - } - - for r in all_recs: - amount = r.price * r.vol * 100.0 # A股 1手=100股 - direction = "in" if r.buyorsell == 0 else "out" if r.buyorsell == 1 else None - if not direction: - continue - - if amount >= 1000000: - stats[f"super_{direction}"] += amount - elif amount >= 200000: - stats[f"large_{direction}"] += amount - elif amount >= 40000: - stats[f"medium_{direction}"] += amount - else: - stats[f"small_{direction}"] += amount - - return FundFlow(**stats) + + return all_recs + + def get_fund_flow(self, market: Market, code: str) -> FundFlow: + """获取个股当日资金流向分布（基于 L1 逐笔数据统计）。""" + records = self._collect_transaction_records( + lambda start, page_size: self.get_transaction_data(market, code, start, page_size), + 2000, + ) + return _classify_fund_flow(records) def get_history_fund_flow( self, market: Market, code: str, start: int, count: int ) -> list[HistoricalFundFlow]: - """获取个股历史日线资金流向序列（Category 22）。 + """获取个股历史日线资金流向序列。 - [EXPERIMENTAL] 当前多台公开主机对该请求仍可能返回空列表。 + 优先走 Category 22 直连接口；若服务器返回空列表，则自动回退为 + “日 K 线取日期 + 历史逐笔成交重算资金流”的兼容实现。 """ - return self._execute(GetHistoryFundFlowCmd(market, code, start, count)) + try: + direct = self._execute(GetHistoryFundFlowCmd(market, code, start, count)) + except Exception: + direct = [] + if direct: + return direct + + bars = self.get_security_bars(market, code, KlineCategory.DAY, start, count) + results: list[HistoricalFundFlow] = [] + for bar in bars: + date = _date_from_bar(bar) + records = self._collect_transaction_records( + lambda page_start, page_size: self.get_history_transaction_data( + market, code, date, page_start, page_size + ), + 800, + ) + results.append(_historical_fund_flow_from_records(date, records)) + return results # ============================================================ @@ -588,6 +696,29 @@ class AsyncTdxClient: ) -> list[SecurityQuote]: return await self._execute(GetSecurityQuotesCmd(stocks)) + async def get_price_limits( + self, market: Market, code: str, name: str, pre_close: float + ) -> tuple[float | None, float | None]: + """按当前交易状态计算涨跌停价。""" + listed_days: int | None = None + no_limit_window_days = get_no_limit_window_days(market, code, name) + if no_limit_window_days > 0: + try: + bars = await self.get_security_bars( + market, code, KlineCategory.DAY, 0, no_limit_window_days + 1 + ) + listed_days = len(bars) + except Exception: + listed_days = None + + return compute_price_limits( + market, + code, + name, + pre_close, + listed_days=listed_days, + ) + async def get_security_bars( self, market: Market, @@ -703,73 +834,83 @@ class AsyncTdxClient: total_volume=q.vol, ) - async def get_fund_flow(self, market: Market, code: str) -> FundFlow: - """获取个股当日资金流向分布（基于 L1 逐笔数据统计）。""" - # 1. 分页拉取当日分笔并去重 + async def _collect_transaction_records( + self, + fetch_page: Callable[[int, int], Awaitable[list[TransactionRecord]]], + page_size: int, + max_start: int = 10000, + ) -> list[TransactionRecord]: all_recs: list[TransactionRecord] = [] - seen_sig = set() - seen_page_sigs = set() + seen_sig: set[tuple[int, int, float, int, int, int]] = set() + seen_page_sigs: set[ + tuple[ + tuple[int, int, float, int, int, int], + tuple[int, int, float, int, int, int], + ] + ] = set() start = 0 - - while start < 10000: - recs = await self.get_transaction_data(market, code, start, 2000) + + while start < max_start: + recs = await fetch_page(start, page_size) if not recs: break - - # 页签名判断：首尾记录组合 - page_sig = ( - ( - recs[0].hour, recs[0].minute, recs[0].price, - recs[0].vol, recs[0].buyorsell, recs[0].unknown_last - ), - ( - recs[-1].hour, recs[-1].minute, recs[-1].price, - recs[-1].vol, recs[-1].buyorsell, recs[-1].unknown_last - ), - ) + + page_sig = _page_signature(recs) if page_sig in seen_page_sigs: break seen_page_sigs.add(page_sig) new_count = 0 - for r in recs: - sig = (r.hour, r.minute, r.price, r.vol, r.buyorsell, r.unknown_last) + for record in recs: + sig = _record_signature(record) if sig not in seen_sig: seen_sig.add(sig) - all_recs.append(r) + all_recs.append(record) new_count += 1 - + if new_count == 0: break - + start += len(recs) if len(recs) < 100: break - - stats = { - "super_in": 0.0, "large_in": 0.0, "medium_in": 0.0, "small_in": 0.0, - "super_out": 0.0, "large_out": 0.0, "medium_out": 0.0, "small_out": 0.0, - } - for r in all_recs: - amount = r.price * r.vol * 100.0 - direction = "in" if r.buyorsell == 0 else "out" if r.buyorsell == 1 else None - if not direction: - continue - if amount >= 1000000: - stats[f"super_{direction}"] += amount - elif amount >= 200000: - stats[f"large_{direction}"] += amount - elif amount >= 40000: - stats[f"medium_{direction}"] += amount - else: - stats[f"small_{direction}"] += amount - return FundFlow(**stats) + + return all_recs + + async def get_fund_flow(self, market: Market, code: str) -> FundFlow: + """获取个股当日资金流向分布（基于 L1 逐笔数据统计）。""" + records = await self._collect_transaction_records( + lambda start, page_size: self.get_transaction_data( + market, code, start, page_size + ), + 2000, + ) + return _classify_fund_flow(records) async def get_history_fund_flow( self, market: Market, code: str, start: int, count: int ) -> list[HistoricalFundFlow]: - """获取个股历史日线资金流向序列（Category 22）。 + """获取个股历史日线资金流向序列。 - [EXPERIMENTAL] 当前多台公开主机对该请求仍可能返回空列表。 + 优先走 Category 22 直连接口；若服务器返回空列表，则自动回退为 + “日 K 线取日期 + 历史逐笔成交重算资金流”的兼容实现。 """ - return await self._execute(GetHistoryFundFlowCmd(market, code, start, count)) + try: + direct = await self._execute(GetHistoryFundFlowCmd(market, code, start, count)) + except Exception: + direct = [] + if direct: + return direct + + bars = await self.get_security_bars(market, code, KlineCategory.DAY, start, count) + results: list[HistoricalFundFlow] = [] + for bar in bars: + date = _date_from_bar(bar) + records = await self._collect_transaction_records( + lambda page_start, page_size: self.get_history_transaction_data( + market, code, date, page_start, page_size + ), + 800, + ) + results.append(_historical_fund_flow_from_records(date, records)) + return results diff --git a/src/xmtdx/codec/price_rules.py b/src/xmtdx/codec/price_rules.py index bebc19b..c2ed4d1 100644 --- a/src/xmtdx/codec/price_rules.py +++ b/src/xmtdx/codec/price_rules.py @@ -4,12 +4,46 @@ from ..models.enums import Market from ..models.finance import FinanceInfo +def get_no_limit_window_days(market: Market, code: str, name: str) -> int: + """返回上市初期不设涨跌幅限制的交易日窗口。 + + 返回值: + 0: 默认按常规涨跌幅限制处理 + 1: 北交所上市首日不设涨跌幅限制 + 5: 沪深主板/创业板/科创板上市前 5 个交易日不设涨跌幅限制 + """ + if _is_index_like(market, code, name): + return 0 + + if code.startswith(("43", "83", "87", "92")): + return 1 + + if market == Market.SH and code.startswith(("60", "68")): + return 5 + if market == Market.SZ and code.startswith(("00", "30")): + return 5 + + return 0 + + +def _is_index_like(market: Market, code: str, name: str) -> bool: + """判断是否为指数/板块类代码。""" + if market == Market.SH and code.startswith( + ("000", "880", "881", "882", "883", "884", "885", "999") + ): + return True + if market == Market.SZ and code.startswith(("395", "399")): + return True + return "指数" in name or "板块" in name + + def compute_price_limits( market: Market, code: str, name: str, pre_close: float, finance_info: FinanceInfo | None = None, + listed_days: int | None = None, ) -> tuple[float | None, float | None]: """根据板块规则计算涨跌停价。 @@ -17,6 +51,11 @@ def compute_price_limits( (limit_up, limit_down) 无涨跌幅限制或当前规则无法可靠判断时返回 ``(None, None)``。 + + Args: + listed_days: + 已上市交易天数（按交易日计，首日=1）。 + 若提供该值，函数会按上市初期无涨跌幅限制规则优先返回 ``(None, None)``。 """ if pre_close <= 0: return None, None @@ -24,18 +63,11 @@ def compute_price_limits( upper_name = name.upper() # 指数/板块类代码通常无涨跌停。 - # 这里优先用明确的指数代码段判断，再用名称兜底，避免把真实股票误判成指数。 - is_index = False - if market == Market.SH and code.startswith( - ("000", "880", "881", "882", "883", "884", "885", "999") - ): - is_index = True - elif market == Market.SZ and code.startswith(("395", "399")): - is_index = True - elif "指数" in name or "板块" in name: - is_index = True + if _is_index_like(market, code, name): + return None, None - if is_index: + no_limit_window_days = get_no_limit_window_days(market, code, name) + if listed_days is not None and 0 < listed_days <= no_limit_window_days: return None, None limit_pct = 0.10 # 默认 10% @@ -50,7 +82,7 @@ def compute_price_limits( elif code.startswith(("43", "83", "87", "92")): limit_pct = 0.30 - # TODO: 上市前 5 日无涨跌幅限制判断（需要 ipo_date 或更明确的上市状态标识）。 + # `listed_days` 是更可靠的交易日维度输入；finance_info 仍保留给上层调用方扩展。 _ = finance_info def _round_price(p: float) -> float: diff --git a/tests/unit/test_a_share_extensions.py b/tests/unit/test_a_share_extensions.py index e746b55..facbf4e 100644 --- a/tests/unit/test_a_share_extensions.py +++ b/tests/unit/test_a_share_extensions.py @@ -4,8 +4,10 @@ import struct from unittest.mock import patch from xmtdx import Market, TdxClient +from xmtdx.models.bar import SecurityBar from xmtdx.models.quote import SecurityQuote from xmtdx.models.security import SecurityInfo +from xmtdx.models.stats import HistoricalFundFlow from xmtdx.models.timeseries import TransactionRecord @@ -115,3 +117,90 @@ def test_get_history_fund_flow_parsing(): assert res[0].year == 2025 assert res[0].month == 1 assert res[0].day == 8 + + +@patch("xmtdx.client.TdxConnection") +def test_get_history_fund_flow_fallback(_mock_conn_cls): + """Category 22 空回包时，自动回退到历史逐笔重算。""" + client = TdxClient("127.0.0.1") + + bars = [ + SecurityBar(10, 10, 10, 10, 0, 0, 2025, 1, 8, 15, 0), + SecurityBar(10, 10, 10, 10, 0, 0, 2025, 1, 9, 15, 0), + ] + txn_map = { + 20250108: [ + TransactionRecord(10, 0, 100.0, 100, 0, 0), + TransactionRecord(10, 1, 10.0, 250, 1, 0), + ], + 20250109: [ + TransactionRecord(10, 0, 10.0, 10, 0, 0), + ], + } + + def mock_history_txn(_market, _code, date, start, count): + if start > 0: + return [] + return txn_map[date] + + with patch.object(TdxClient, "_execute", return_value=[]), patch.object( + TdxClient, "get_security_bars", return_value=bars + ), patch.object( + TdxClient, "get_history_transaction_data", side_effect=mock_history_txn + ): + flows = client.get_history_fund_flow(Market.SH, "600000", 0, 2) + + assert flows == [ + HistoricalFundFlow( + year=2025, + month=1, + day=8, + super_in=1000000.0, + super_out=0.0, + large_in=0.0, + large_out=250000.0, + medium_in=0.0, + medium_out=0.0, + small_in=0.0, + small_out=0.0, + ), + HistoricalFundFlow( + year=2025, + month=1, + day=9, + super_in=0.0, + super_out=0.0, + large_in=0.0, + large_out=0.0, + medium_in=0.0, + medium_out=0.0, + small_in=10000.0, + small_out=0.0, + ), + ] + + +@patch("xmtdx.client.TdxConnection") +def test_get_price_limits_uses_listing_window(_mock_conn_cls): + """client.get_price_limits 应结合日 K 条数判断上市初期限价窗口。""" + client = TdxClient("127.0.0.1") + + with patch.object( + TdxClient, + "get_security_bars", + return_value=[SecurityBar(0, 0, 0, 0, 0, 0, 2025, 1, 1, 15, 0)] * 5, + ): + assert client.get_price_limits(Market.SH, "600001", "主板新股", 10.0) == ( + None, + None, + ) + + with patch.object( + TdxClient, + "get_security_bars", + return_value=[SecurityBar(0, 0, 0, 0, 0, 0, 2025, 1, 1, 15, 0)] * 6, + ): + assert client.get_price_limits(Market.SH, "600001", "主板老股", 10.0) == ( + 11.0, + 9.0, + ) diff --git a/tests/unit/test_protocol_fixes.py b/tests/unit/test_protocol_fixes.py index 54dbe28..b13824c 100644 --- a/tests/unit/test_protocol_fixes.py +++ b/tests/unit/test_protocol_fixes.py @@ -107,3 +107,19 @@ def test_compute_price_limits_for_indices(): assert compute_price_limits(Market.SH, "999999", "上证指数", 4026.63) == (None, None) assert compute_price_limits(Market.SH, "880005", "涨跌家数", 1841.0) == (None, None) assert compute_price_limits(Market.SZ, "399001", "深证成指", 10412.63) == (None, None) + + +def test_compute_price_limits_for_newly_listed_stocks(): + """上市初期限价窗口应返回 None。""" + assert compute_price_limits( + Market.SH, "600001", "主板新股", 10.0, listed_days=5 + ) == (None, None) + assert compute_price_limits( + Market.SH, "600001", "主板新股", 10.0, listed_days=6 + ) == (11.0, 9.0) + assert compute_price_limits( + Market.BJ, "920002", "北交所新股", 84.36, listed_days=1 + ) == (None, None) + assert compute_price_limits( + Market.BJ, "920002", "北交所新股", 84.36, listed_days=2 + ) == (109.67, 59.05)