diff --git a/src/xmtdx/client.py b/src/xmtdx/client.py index 402b797..2a3bae9 100644 --- a/src/xmtdx/client.py +++ b/src/xmtdx/client.py @@ -75,11 +75,11 @@ def _classify_fund_flow(records: list[TransactionRecord]) -> FundFlow: if not direction: continue - if amount >= 1_000_000: + if amount > 1_000_000: stats[f"super_{direction}"] += amount - elif amount >= 200_000: + elif amount > 200_000: stats[f"large_{direction}"] += amount - elif amount >= 40_000: + elif amount > 40_000: stats[f"medium_{direction}"] += amount else: stats[f"small_{direction}"] += amount diff --git a/src/xmtdx/commands/fund_flow.py b/src/xmtdx/commands/fund_flow.py index b0437db..40279d9 100644 --- a/src/xmtdx/commands/fund_flow.py +++ b/src/xmtdx/commands/fund_flow.py @@ -52,8 +52,7 @@ class GetHistoryFundFlowCmd(BaseCommand[list[HistoricalFundFlow]]): # 记录格式：4字节日期 + 8个4字节自定义浮点金额 # [0]日期, [1..4]流入(超/大/中/小), [5..8]流出(超/大/中/小) - # 这里的 i 是 uint32 原始字节，随后需用 _decode_volume 解码 - raw_data = struct.unpack(" str: """将 reversed_bytes0 整数转换为 HH:MM:SS.mmm 字符串。 - 方法来自 pytdx issue #187。raw 为 14999212 → "14:59:57.163" + 该字段编码为“小时 + 百万分之一小时的小数部分”。 + 例如：14999212 → "14:59:57.163" """ - s = str(raw) - if len(s) < 6: - return s - # 最后6位：前两位=秒，后四位=毫秒的某种编码 - time_part = s[:-6] + ":" - last6 = int(s[-6:]) - if int(s[-6:-4]) < 60: - time_part += s[-6:-4] + ":" - time_part += f"{last6 % 10000 * 60 / 10000.0:06.3f}" - else: - mins = last6 * 60 // 1000000 - secs = (last6 * 60 % 1000000) * 60 / 1000000.0 - time_part += f"{mins:02d}:{secs:06.3f}" - return time_part + hours, fractional_hour = divmod(raw, 1_000_000) + total_millis = fractional_hour * 3600 // 1000 + minutes, remainder = divmod(total_millis, 60_000) + seconds, millis = divmod(remainder, 1000) + return f"{hours:02d}:{minutes:02d}:{seconds:02d}.{millis:03d}" class GetSecurityQuotesCmd(BaseCommand[list[SecurityQuote]]): diff --git a/src/xmtdx/models/stats.py b/src/xmtdx/models/stats.py index a02267f..060edba 100644 --- a/src/xmtdx/models/stats.py +++ b/src/xmtdx/models/stats.py @@ -20,9 +20,9 @@ class FundFlow: # 流入项 (Buy) super_in: float # 超大单流入 (>100万) - large_in: float # 大单流入 (20万-100万) - medium_in: float # 中单流入 (4万-20万) - small_in: float # 小单流入 (<4万) + large_in: float # 大单流入 (>20万 且 <=100万) + medium_in: float # 中单流入 (>4万 且 <=20万) + small_in: float # 小单流入 (<=4万) # 流出项 (Sell) super_out: float diff --git a/src/xmtdx/transport/sync.py b/src/xmtdx/transport/sync.py index f86dd52..72dec4c 100644 --- a/src/xmtdx/transport/sync.py +++ b/src/xmtdx/transport/sync.py @@ -126,7 +126,15 @@ class TdxConnection: sock.close() raise TdxConnectionError(f"无法连接 {self.host}:{self.port}: {e}") from e self._sock = sock - self._send_setup() + try: + self._send_setup() + except Exception: + try: + sock.close() + except OSError: + pass + self._sock = None + raise def close(self) -> None: """关闭连接。""" diff --git a/tests/unit/test_a_share_extensions.py b/tests/unit/test_a_share_extensions.py index facbf4e..9b9dcd3 100644 --- a/tests/unit/test_a_share_extensions.py +++ b/tests/unit/test_a_share_extensions.py @@ -4,6 +4,7 @@ import struct from unittest.mock import patch from xmtdx import Market, TdxClient +from xmtdx.client import _classify_fund_flow from xmtdx.models.bar import SecurityBar from xmtdx.models.quote import SecurityQuote from xmtdx.models.security import SecurityInfo @@ -18,7 +19,7 @@ def test_get_fund_flow_logic(_mock_conn_cls): # 构造模拟 Tick 数据 mock_recs = [ - TransactionRecord(10, 0, 100.0, 100, 0, 0), # super_in (100*100*100 = 100w) + TransactionRecord(10, 0, 100.0, 101, 0, 0), # super_in (100*101*100 = 101w) TransactionRecord(10, 1, 10.0, 250, 1, 0), # large_out (10*250*100 = 25w) TransactionRecord(10, 2, 10.0, 10, 0, 0), # small_in (10*10*100 = 1w) ] @@ -26,10 +27,24 @@ def test_get_fund_flow_logic(_mock_conn_cls): with patch.object(TdxClient, "get_transaction_data", return_value=mock_recs): flow = client.get_fund_flow(Market.SH, "600000") - assert flow.super_in == 1000000.0 + assert flow.super_in == 1010000.0 assert flow.large_out == 250000.0 assert flow.small_in == 10000.0 - assert flow.main_net_inflow == 1000000.0 - 250000.0 + assert flow.main_net_inflow == 1010000.0 - 250000.0 + + +def test_classify_fund_flow_exact_thresholds_use_lower_bucket(): + """恰好命中阈值时，应落入较低一档。""" + flow = _classify_fund_flow([ + TransactionRecord(10, 0, 100.0, 100, 0, 0), # 100w -> large + TransactionRecord(10, 1, 100.0, 20, 0, 0), # 20w -> medium + TransactionRecord(10, 2, 100.0, 4, 0, 0), # 4w -> small + ]) + + assert flow.super_in == 0.0 + assert flow.large_in == 1000000.0 + assert flow.medium_in == 200000.0 + assert flow.small_in == 40000.0 @patch("xmtdx.client.TdxConnection") def test_get_security_list_all_filtering(_mock_conn_cls): @@ -103,11 +118,11 @@ def test_get_history_fund_flow_parsing(): body = bytearray(9) body.extend(struct.pack(" float: + seen.append(raw) + return float(raw) + + with patch("xmtdx.commands.fund_flow._decode_volume", side_effect=fake_decode): + records = GetHistoryFundFlowCmd(Market.SH, "600000", 0, 1).parse_response(bytes(body)) + + assert seen == raw_words + assert records[0].small_out == float(raw_words[-1]) diff --git a/tests/unit/test_sync_transport.py b/tests/unit/test_sync_transport.py new file mode 100644 index 0000000..6f2daa7 --- /dev/null +++ b/tests/unit/test_sync_transport.py @@ -0,0 +1,44 @@ +"""同步 transport 回归测试。""" + +from unittest.mock import patch + +from xmtdx.exceptions import TdxConnectionError +from xmtdx.transport.sync import TdxConnection + + +class _FakeSocket: + def __init__(self) -> None: + self.timeout: float | None = None + self.connected_to: tuple[str, int] | None = None + self.closed = False + + def settimeout(self, timeout: float) -> None: + self.timeout = timeout + + def connect(self, address: tuple[str, int]) -> None: + self.connected_to = address + + def close(self) -> None: + self.closed = True + + +def test_sync_connection_closes_socket_when_setup_fails() -> None: + sock = _FakeSocket() + conn = TdxConnection("127.0.0.1", port=7709, timeout=0.2) + + with patch("xmtdx.transport.sync.socket.socket", return_value=sock), patch.object( + TdxConnection, + "_send_setup", + side_effect=TdxConnectionError("setup failed"), + ): + try: + conn.connect() + except TdxConnectionError as exc: + assert "setup failed" in str(exc) + else: # pragma: no cover - 防御性断言 + raise AssertionError("expected setup failure") + + assert sock.timeout == 0.2 + assert sock.connected_to == ("127.0.0.1", 7709) + assert sock.closed is True + assert conn._sock is None