rs-czsc实现中check_fxs 产生大量 "分型时间相同" 警告

[ohydo]

Bug Report: check_fxs 产生大量 "分型时间相同" 警告

问题描述

在使用 rs-czsc (Rust 实现) 处理60分钟K线数据时，check_fxs 函数产生大量警告，提示分型时间相同。但在相同数据上使用 Python 实现 (CZSC_USE_PYTHON=1) 时，警告消失。

复现步骤

测试数据

• 60分钟K线数据
• 时间精度：分钟级别（无秒信息）
• 数据时间唯一，无重复

复现代码

import os
import warnings
import pandas as pd
from datetime import datetime
# 使用 Rust 实现（默认）
# os.environ["CZSC_USE_PYTHON"] = "0"  # 或者不设置
from czsc import CZSC
# 加载60分钟K线数据
bars = [...]  # RawBar 列表，时间格式如 2021-12-20 15:00:00
with warnings.catch_warnings(record=True) as w:
    warnings.simplefilter("always")
    
    c = CZSC(bars)
    
    # 逐根K线更新
    for bar in bars[500:]:  # 预热后开始
        c.update(bar)
        
    # 检查警告
    check_fxs_warnings = [x for x in w if "check_fxs" in str(x.message)]
    print(f"check_fxs 警告数量: {len(check_fxs_warnings)}")
    
    # 打印前几个警告
    for warning in check_fxs_warnings[:5]:
        print(f"  {warning.message}")
实际输出（Rust 实现）
check_fxs 警告数量: 100+ (大量)
  check_fxs错误: 2021-12-20 15:00:00 UTC，G，G
  check_fxs错误: 2021-12-21 10:00:00 UTC，G，G
  ...
警告格式：check_fxs错误: <时间> UTC，<分型类型1>，<分型类型2>
关键特征：两个分型类型相同（如 "G，G"），表示检测到两个相同类型的分型在同一时间。
对比输出（Python 实现）
os.environ["CZSC_USE_PYTHON"] = "1"  # 强制使用 Python 实现
# 相同代码...
输出：
check_fxs 警告数量: 0
预期行为
check_fxs 不应产生大量 "分型时间相同" 警告，因为：
1. 输入的60分钟K线时间唯一，无重复
2. Python 实现处理相同数据无警告
3. 分型时间相同（如 "G，G"）不应出现，因为相邻分型类型应该交替（顶分型后应为底分型）
可能原因分析
1. 时间精度处理问题
60分钟K线时间格式为 2021-12-20 15:00:00（无秒信息）。
Python 实现 (czsc/py/analyze.py 第107-133行)：
def check_fxs(fxs: List[FX]) -> None:
    """检查分型列表是否有效"""
    for i in range(1, len(fxs)):
        fx1, fx2 = fxs[i - 1], fxs[i]
        if fx1.dt == fx2.dt:
            warnings.warn(f"check_fxs错误：{fx1.dt}，{fx1.mark}，{fx2.mark}")
Python 实现简单比较 fx1.dt == fx2.dt。
Rust 实现 可能：
- 时间比较精度不一致
- 或分型时间生成逻辑有差异
2. 分型生成逻辑差异
分型时间 (fx.dt) 通常取分型中间K线的时间。
60分钟数据无秒精度时：
- 分型时间精度：分钟级
- 如果相邻分型的中间K线在同一分钟，可能导致时间相同
但这种情况理论上不应频繁出现，且 Python 实现未出现此问题。
3. 时区处理
警告显示时间带有 UTC 后缀，可能存在时区转换问题。
环境信息
项目	信息
Python 版本	3.10+
rs-czsc 版本	(获取不了具体版本，只显示未知)
操作系统	Linux
CZSC 版本	0.10.x
临时解决方案
在 Python 代码中设置环境变量强制使用 Python 实现：
import os
os.environ["CZSC_USE_PYTHON"] = "1"  # 必须在导入 czsc 之前
或使用 spawn 启动方法的子进程：
import multiprocessing
import pickle
def worker(engine_pickle, symbol):
    os.environ["CZSC_USE_PYTHON"] = "1"
    engine = pickle.loads(engine_pickle)
    return engine.run_symbol(symbol)
ctx = multiprocessing.get_context("spawn")
with ProcessPoolExecutor(mp_context=ctx) as executor:
    results = executor.submit(worker, pickle.dumps(engine), symbol)
相关代码位置
- Python check_fxs 实现：czsc/py/analyze.py 第107-133行
- Rust check_fxs 实现：rs-czsc 库内部
请求
1. 请确认 Rust 版 check_fxs 的时间比较逻辑
2. 请检查分型时间生成逻辑与 Python 版本的一致性
3. 如有可能，请提供修复版本或指导如何调试 Rust 代码
附：Python check_fxs 实现参考
def check_fxs(fxs: List[FX]) -> None:
    """检查分型列表
    
    分型列表应该满足：
    1. 两个相邻分型的时间不能相同
    2. 两个相邻分型的类型必须相反（顶分型后必须是底分型）
    
    :param fxs: 分型列表
    """
    if len(fxs) == 0:
        return
    
    for i in range(1, len(fxs)):
        fx1, fx2 = fxs[i - 1], fxs[i]
        
        # 检查时间
        if fx1.dt == fx2.dt:
            warnings.warn(f"check_fxs错误：{fx1.dt}，{fx1.mark}，{fx2.mark}")
        
        # 检查类型交替
        if fx1.mark == fx2.mark:
            warnings.warn(f"check_fxs错误：{fx1.dt}，{fx1.mark}，{fx2.mark}")
---
感谢维护者的辛勤工作！