Add live fallbacks for history fund flow and price limits

2026-09-12 14:34:18 +08:00 · 2026-04-15 21:23:13 +08:00
parent 96f14f70bc
commit 4cb48daabb
6 changed files with 402 additions and 121 deletions
@@ -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（分时）

@@ -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):")
@@ -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
@@ -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:
@@ -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,
+        )
@@ -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)