technical-patterns-lab/discuss/20260130-生产环境性能评估_每日5000股.md

生产环境性能评估：每日5000股检测场景

使用场景

真实需求：

• 股票数量：5000+只
• 运行频率：每天执行一次
• 检测参数：可自定义（窗口、收敛度、突破阈值等）
• 输出要求：标准化强度分 [0,1]

---

性能瓶颈分析

完整流程耗时分解

┌─────────────────────────────────────────────────────────────┐
│ 每日检测流程                                                │
├─────────────────────────────────────────────────────────────┤
│ 1. 数据加载 (load_pkl)          ~0.5-2秒   (I/O瓶颈)      │
│ 2. 三角形批量检测                ~10-60秒   (计算密集)     │
│ 3. 标准化处理 (7个维度)         ~0.05-0.2秒 (CPU)         │
│ 4. 强度分计算 (加权求和)        ~0.01秒     (向量运算)    │
│ 5. 结果输出/存储                 ~0.1-0.5秒  (I/O)         │
├─────────────────────────────────────────────────────────────┤
│ 总耗时：~11-63秒/天                                         │
└─────────────────────────────────────────────────────────────┘

结论：标准化不是瓶颈！

• 标准化耗时 < 0.5% 总时间
• 真正瓶颈：三角形检测算法 (占90%+)

---

关键问题：标准化基准如何选择？

场景A：基于历史基准（18,004样本）

# 预先计算并缓存历史基准
HISTORICAL_QUANTILES = {
    'price_score_up': {'p5': 0.0, 'p50': 0.0, 'p95': 0.15},
    'convergence_score': {'p5': 0.45, 'p50': 0.80, 'p95': 0.92},
    # ... 其他维度
}

def normalize_with_historical_baseline(today_data):
    """每天用历史基准标准化今天的5000只股票"""
    for col in columns:
        p5, p50, p95 = HISTORICAL_QUANTILES[col].values()
        normalized = apply_normalization(today_data[col], p5, p50, p95)
    return normalized

优点：

• ✅ 标准化基准稳定，不随每日数据波动
• ✅ 预设模式阈值(0.7/0.75)仍然有效
• ✅ 不同日期的强度分可直接对比
• ✅ 每天只需O(n)线性映射，极快

缺点：

• ❌ 需要定期更新历史基准（如每季度）
• ❌ 极端市场环境下可能出现大量值超出[0,1]范围（需clip）

---

场景B：基于当天样本（5000只）

def normalize_with_daily_baseline(today_data):
    """每天用当天5000只股票重新计算分位数"""
    for col in columns:
        p5 = today_data[col].quantile(0.05)
        p50 = today_data[col].median()
        p95 = today_data[col].quantile(0.95)
        normalized = apply_normalization(today_data[col], p5, p50, p95)
    return normalized

优点：

• ✅ 自适应市场环境，无需维护历史基准
• ✅ 保证当天数据完全在[0,1]范围

缺点：

• ❌ 标准化基准每天变化，不同日期强度分不可比
• ❌ 预设模式阈值失效（今天0.7 ≠ 明天0.7）
• ❌ 牛市/熊市时相对排名被压缩

---

推荐方案：混合策略

方案：历史基准 + 快速线性映射

# ========== 离线预计算（每季度更新一次） ==========
def compute_historical_baseline(historical_data_18004_samples):
    """
    基于历史数据计算标准化基准
    输入：18,004个历史样本
    输出：每个维度的P5/P50/P95
    """
    baseline = {}
    for col in ['price_score_up', 'price_score_down', ...]:
        baseline[col] = {
            'p5': historical_data[col].quantile(0.05),
            'p50': historical_data[col].median(),
            'p95': historical_data[col].quantile(0.95),
            'method': 'zero_inflated' if col in [...] else 'standard'
        }
    
    # 保存为配置文件
    save_json('baseline_config.json', baseline)
    return baseline


# ========== 在线标准化（每天运行） ==========
def normalize_daily_fast(today_5000_samples, baseline_config):
    """
    使用历史基准快速标准化今天的数据
    
    时间复杂度：O(n) = O(5000) ≈ 10-20ms
    vs 百分位排名 O(n log n) ≈ 50-100ms
    """
    result = {}
    
    for col, config in baseline_config.items():
        p5 = config['p5']
        p50 = config['p50']
        p95 = config['p95']
        method = config['method']
        
        if method == 'zero_inflated':
            # 零膨胀：零值→0.5，非零值线性映射到[0.5, 1.0]
            is_zero = (today_5000_samples[col] < 1e-6)
            result[col] = np.where(
                is_zero, 
                0.5,
                0.5 + 0.5 * ((today_5000_samples[col] - p5) / (p95 - p5)).clip(0, 1)
            )
        elif method == 'median_aligned':
            # 中位数对齐：上半[p50,p95]→[0.5,1.0]，下半[p5,p50]→[0.0,0.5]
            is_upper = (today_5000_samples[col] >= p50)
            upper_norm = 0.5 + 0.5 * ((today_5000_samples[col] - p50) / (p95 - p50)).clip(0, 1)
            lower_norm = 0.5 * ((today_5000_samples[col] - p5) / (p50 - p5)).clip(0, 1)
            result[col] = np.where(is_upper, upper_norm, lower_norm)
        else:
            # 标准线性映射
            result[col] = ((today_5000_samples[col] - p5) / (p95 - p5)).clip(0, 1)
    
    return pd.DataFrame(result)

---

性能对比（5000样本/天）

方法	耗时	相比当前	质量	复杂度
当前：rank(pct=True)	50-100ms	基准	⭐⭐⭐⭐⭐	简单
历史基准+线性映射	10-20ms	5倍提速	⭐⭐⭐⭐	中等
每日重算baseline	15-30ms	3倍提速	⭐⭐⭐	中等

结论：

• 对于5000样本，优化收益从 50ms → 20ms，节省30ms
• 相对于检测算法的10-60秒，收益微乎其微 (<0.1%)
• 不建议优化标准化环节，应优化检测算法

---

实际生产建议

短期（保持现状）

def 收敛三角形_生产环境(today_data_5000_stocks):
    """
    每天运行，使用历史基准标准化
    """
    # 1. 加载历史基准（只需加载一次，常驻内存）
    baseline = load_cached_baseline()
    
    # 2. 检测三角形（主要耗时：10-60秒）
    detection_result = detect_converging_triangle_batch(today_data_5000_stocks)
    
    # 3. 使用历史基准标准化（耗时：10-20ms，用线性映射）
    normalized = normalize_with_historical_baseline(
        detection_result, 
        baseline
    )
    
    # 4. 计算强度分（耗时：<10ms）
    strength = calculate_strength(normalized, CONFIG_EQUAL)
    
    return strength

关键点：

1. 历史基准常驻内存，不用每天重算
2. 标准化用线性映射（历史基准+快速计算）
3. 优化重点：检测算法，而非标准化

---

中期（如果检测算法已优化）

场景：检测算法优化到1-2秒后，标准化才可能成为瓶颈

此时可考虑：

1. 预编译Numba/Cython加速标准化
2. 多进程并行化（5000股分成10组并行）
3. GPU加速（如果数据量进一步增大）

---

长期（大规模扩展）

场景：股票数量 > 1万只，或需要分钟级实时计算

# 使用增量更新策略
class IncrementalNormalizer:
    """增量标准化器，维护滚动窗口基准"""
    
    def __init__(self, window_size=20_days):
        self.historical_buffer = deque(maxlen=window_size)
        self.baseline = None
    
    def update(self, today_data):
        """每天更新基准"""
        self.historical_buffer.append(today_data)
        
        # 重新计算最近20天的基准（采样策略）
        if len(self.historical_buffer) == self.window_size:
            self.baseline = compute_baseline_fast(self.historical_buffer)
    
    def normalize(self, today_data):
        """O(1)标准化"""
        return apply_linear_mapping(today_data, self.baseline)

---

决策树

是否需要优化标准化？
│
├─ 检测算法耗时 > 10秒？
│  ├─ 是 → 先优化检测算法（收益更大）
│  └─ 否 → 继续往下
│
├─ 标准化耗时 > 500ms？
│  ├─ 是 → 考虑优化（使用历史基准+线性映射）
│  └─ 否 → 不需要优化
│
└─ 总结：对于5000样本/天，标准化不是瓶颈

---

实施建议

立即行动

1. 基准测试：实测当前5000股的实际耗时

   python scripts/benchmark_production.py --stocks 5000
   

2. 分析瓶颈：确认检测算法 vs 标准化的耗时占比

3. 优先级排序：

   • P0: 优化检测算法（如果>10秒）
   • P1: 优化数据加载I/O（如果>2秒）
   • P2: 优化标准化（如果>500ms）

如果确需优化标准化

步骤：

1. 离线计算历史基准（每季度更新）
2. 修改标准化函数使用线性映射
3. A/B测试质量影响
4. 监控每日强度分分布

代码位置：

dunhe_dataServer/src/library/expression/funcs/pattern.py
- 修改 _compute_all_metrics() 中的标准化逻辑
- 增加 load_historical_baseline() 函数
- 增加 normalize_with_baseline_fast() 函数

---

结论

对于每天5000只股票的场景：

1. ✅ 保持当前方案（百分位排名）

   • 标准化耗时 < 100ms，不是瓶颈
   • 质量最优，稳定可靠

2. 📊 如果未来需要优化（检测算法已优化到秒级）

   • 使用历史基准 + 线性映射
   • 5倍提速（100ms → 20ms）
   • 质量略降（0.98 → 0.88）

3. 🎯 真正应该优化的：

   • 检测算法性能（占90%+时间）
   • 数据加载I/O
   • 结果缓存策略

一句话总结：标准化优化对每天5000股场景的收益 < 0.1%，不值得投入。应聚焦检测算法优化。