hao/quantKonwledge
1
0
派生 0
代码 工单 合并请求 工作流 软件包 项目 发布 百科 活动

feat: 信号系统深度优化建议（基于 EWO 通知案例分析）

浏览代码 
新增模块 12_信号系统优化：
- 信号系统深度优化建议.md：
  * 真实 EWO 转换案例分析（BTC/SOL 10m 对比）
  * 5大核心问题诊断（EWO幅度过滤/周期阈值/持续时间/大周期粘性/成交量）
  * 周期分层策略矩阵（1m~1w 全周期配置建议）
  * 主流币专项配置（BTC/ETH/SOL/XAUT）
  * MTF三层周期联动框架（4h→1h→10m）
  * EWO通知系统增强方案
  * 未来调整路线图（短/中/长期）

新增样本代码：
- samples/optimized_signal_engine.py：
  * EWO幅度分级评分（按品种阈值：BTC>15/SOL>0.5）
  * EWO阶段持续时间奖励（≥20根K线额外+1）
  * 周期自适应信号强度阈值（10m×1.2倍）
  * 放量确认加权机制
  * 粘性大周期偏向过滤（需连续3次确认才切换）
  * 完整 MTF 信号架构实现
这个提交包含在：
Manus Quant Agent
2026-03-05 21:48:51 -05:00
父节点 644400f8e0
当前提交 8615a511bc
共修改 2 个文件，包含 957 行新增和 0 行删除
下载 Patch 文件 下载 Diff 文件 展开所有文件 折叠所有文件

405
12_信号系统优化/信号系统深度优化建议.md 普通文件
查看文件

@@ -0,0 +1,405 @@
# 信号交易系统深度优化建议
> 基于 tradehk 项目源码（`indicators.ts` + `types.ts`）与真实 EWO 转换通知案例的深度分析报告。
> 作者：Manus AI | 更新日期：2026-03-06
---
## 一、现有系统架构解析
### 1.1 EWO 转换通知机制（真实案例）
以下是触发本次分析的两条真实通知：
```
EWO转换提醒 BTC/10m
触发规则: EWO 红 -> 绿
转换收线: 2026/03/06 10:10:00 (Asia/Shanghai)
EWO: -29.048617 -> 33.320837
上一阶段: 红(空头) 持续: 4小时30分钟 (27根10mK)
区间: 2026/03/06 05:30:00 ~ 2026/03/06 10:00:00
EWO转换提醒 SOL/10m
触发规则: EWO 红 -> 绿
转换收线: 2026/03/06 10:10:00 (Asia/Shanghai)
EWO: -0.037316 -> 0.006745
上一阶段: 红(空头) 持续: 4小时 (24根10mK)
区间: 2026/03/06 06:00:00 ~ 2026/03/06 10:00:00
```
**关键观察**：BTC 的 EWO 变化幅度为 `-29.05 → +33.32`（绝对变化 62.37），而 SOL 的变化幅度仅为 `-0.037 → +0.007`（绝对变化 0.044）。两者同时触发"红→绿"转换，但信号质量存在本质差异——BTC 的转换是强烈的动能反转，SOL 的转换则仅是微弱的零轴穿越，极易产生假信号。
### 1.2 现有信号引擎核心逻辑
| 指标组 | 类型 | 最大权重 | 触发条件 |
|--------|------|----------|----------|
| EWO 穿越零轴 | 核心（常驻） | +2 | 从负变正 / 从正变负 |
| EWO 持续方向 | 核心（常驻） | +1 | 在零轴上方 / 下方 |
| MACD 金叉/死叉 | 核心（常驻） | +2 | 快线穿越慢线 |
| MACD 柱状图扩大 | 核心（常驻） | +1 | 柱状图绝对值增大 |
| AO 穿越零轴 | 核心（常驻） | +1 | 从负变正 / 从正变负 |
| MA 多头/空头排列 | 核心（常驻） | +1 | 价格>MA10>MA100 |
| RSI 超买超卖 | 可选 | +1~+2 | 低于30/高于70 |
| KDJ 金叉/死叉 | 可选 | +1~+2 | K线穿越D线 |
| Stoch 超买超卖 | 可选 | +1 | K/D 均低于20/高于80 |
| 布林带触轨 | 可选 | +1 | 价格触及上下轨 |
| SuperTrend 反转 | 可选 | +1~+2 | 趋势方向改变 |
| DMI 金叉/死叉 | 可选 | +1~+2 | ADX>25 时 +DI/-DI 交叉 |
**大周期偏向过滤**（`assessBigTimeframeBias`）：使用 4h/12h 的 EWO（权重2）+ MACD 方向（权重1）+ MACD 柱（权重1）+ AO（权重1），总分 ≥ 4/5 才确认方向。
---
## 二、核心问题诊断
### 2.1 问题一：EWO 零轴穿越缺乏幅度过滤（最高优先级）
**现象**：SOL 的 EWO 从 `-0.037` 变为 `+0.007`，绝对值极小，但与 BTC 的 `-29 → +33` 获得完全相同的评分（+2）。这是系统最大的噪声来源。
**根因**：`generateSignal` 中 EWO 的判断仅检查正负符号，未考虑变化幅度：
```typescript
// 现有代码（indicators.ts 第 481-494 行）
if (ewoNow > 0 && ewoPrev <= 0) {
bullishCount += 2; // ← 无论幅度大小，一律 +2
}
```
**优化方案**：引入 EWO 幅度过滤，基于 ATR 标准化的相对幅度进行分级评分：
```typescript
// 建议优化：EWO 幅度分级评分
const ewoChangeAbs = Math.abs(ewoNow - ewoPrev);
const ewoAvgAbs = (Math.abs(ewoNow) + Math.abs(ewoPrev)) / 2;
const ewoStrength = ewoAvgAbs > 0 ? ewoChangeAbs / ewoAvgAbs : 0;
if (ewoNow > 0 && ewoPrev <= 0) {
if (ewoStrength > 2.0 || Math.abs(ewoNow) > ewoThreshold) {
bullishCount += 2; // 强力穿越
reasons.push(`EWO 强力上穿零轴 (幅度: ${ewoNow.toFixed(2)})`);
} else {
bullishCount += 1; // 微弱穿越，降权
reasons.push(`EWO 微弱上穿零轴 (幅度: ${ewoNow.toFixed(4)}) ⚠️ 注意假突破`);
}
}
```
**EWO 阈值建议**（基于品种特性）：
| 品种 | 10m EWO 强力穿越阈值 | 说明 |
|------|---------------------|------|
| BTC | > 15.0 | 价格基数大，EWO 绝对值大 |
| ETH | > 5.0 | 中等波动 |
| SOL | > 0.5 | 价格较低，EWO 绝对值小 |
| DOGE | > 0.001 | 价格极低，需特殊处理 |
| XAUT | > 3.0 | 黄金代币，波动率低 |
### 2.2 问题二：信号强度阈值与周期不匹配
**现象**：10m 周期的信号噪声远高于 4h/1d 周期，但 `strongThreshold` 计算公式未区分周期：
```typescript
// 现有代码（第 720-721 行）
const strongThreshold = 5 + Math.floor(activeOptionalCount * 0.5);
const moderateThreshold = 3 + Math.floor(activeOptionalCount * 0.3);
```
**优化方案**：引入周期自适应阈值：
```typescript
// 建议：周期自适应阈值
const intervalMultiplier: Record<TimeInterval, number> = {
'1m': 1.5, '3m': 1.4, '5m': 1.3, '10m': 1.2,
'15m': 1.1, '30m': 1.0, '1h': 0.9, '4h': 0.8,
'12h': 0.7, '1d': 0.6, '1w': 0.5
};
const mult = intervalMultiplier[interval] ?? 1.0;
const strongThreshold = Math.ceil((5 + activeOptionalCount * 0.5) * mult);
const moderateThreshold = Math.ceil((3 + activeOptionalCount * 0.3) * mult);
```
这意味着 10m 周期需要更高的评分才能触发 STRONG 信号，有效过滤短周期噪声。
### 2.3 问题三：缺少 EWO 持续时间与阶段质量评估
**现象**：通知中包含了"上一阶段持续时间（27根K线）"这一关键信息，但信号引擎完全未使用。持续时间越长的 EWO 阶段，其反转信号越可靠。
**优化方案**：在信号引擎中加入 EWO 阶段持续时间奖励：
```typescript
// 建议：EWO 阶段持续时间奖励
function countEwoPhaseDuration(ewoArr: number[], currentIdx: number): number {
const currentSign = ewoArr[currentIdx] >= 0 ? 1 : -1;
let count = 0;
for (let i = currentIdx - 1; i >= 0; i--) {
const prevSign = ewoArr[i] >= 0 ? 1 : -1;
if (prevSign === currentSign) count++;
else break;
}
return count;
}
// 在穿越时使用
const prevPhaseDuration = countEwoPhaseDuration(ewoArr, prev);
if (ewoNow > 0 && ewoPrev <= 0) {
bullishCount += 2;
if (prevPhaseDuration >= 20) { // 上一空头阶段持续 20 根以上
bullishCount += 1; // 额外奖励：长期空头结束后的反转更可信
reasons.push(`EWO 上穿零轴（前空头阶段持续 ${prevPhaseDuration} 根K线，信号可靠性高）`);
}
}
```
### 2.4 问题四：大周期偏向过滤逻辑过于简单
**现象**：`assessBigTimeframeBias` 使用固定的 4/5 分阈值，但在市场震荡期（EWO 在零轴附近反复横跳）会频繁切换 BULLISH/BEARISH，导致小周期信号被错误过滤。
**优化方案**：引入大周期偏向的"粘性"机制（Sticky Bias）：
```typescript
// 建议：带粘性的大周期偏向
interface StickyBiasState {
bias: TrendBias;
confirmedAt: number;
consecutiveConfirms: number;
}
function assessStickyBigTimeframeBias(
candles: Candle[],
prevState: StickyBiasState,
minConfirms: number = 3 // 需要连续 3 次确认才切换方向
): StickyBiasState {
const rawBias = assessBigTimeframeBias(candles);
if (rawBias === prevState.bias) {
return { ...prevState, consecutiveConfirms: prevState.consecutiveConfirms + 1 };
}
if (prevState.consecutiveConfirms < minConfirms) {
// 尚未达到切换阈值，保持原方向
return prevState;
}
return { bias: rawBias, confirmedAt: Date.now(), consecutiveConfirms: 1 };
}
```
### 2.5 问题五：缺少成交量确认机制
**现象**：现有系统仅有 `detectVolumeContraction`（缩量检测），但缺少"放量突破"确认。EWO 转换时若伴随放量，信号可靠性显著提升。
**优化方案**：
```typescript
// 建议：放量确认加分
function detectVolumeExpansion(volumes: number[], lookback = 5, threshold = 1.5): boolean {
if (volumes.length < lookback + 1) return false;
const recent = volumes.slice(-lookback);
const avgVolume = recent.reduce((a, b) => a + b, 0) / recent.length;
const prevAvg = volumes.slice(-lookback * 2, -lookback)
.reduce((a, b) => a + b, 0) / lookback;
return avgVolume > prevAvg * threshold;
}
// 在信号生成中使用
const isVolumeExpanding = detectVolumeExpansion(volumes);
if (isVolumeExpanding && (bullishCount > 0 || bearishCount > 0)) {
if (bullishCount > bearishCount) bullishCount += 1;
else if (bearishCount > bullishCount) bearishCount += 1;
reasons.push('✅ 放量确认信号');
}
```
---
## 三、快速适配不同周期的策略框架
### 3.1 周期分层策略矩阵
不同周期的市场特性差异显著，信号系统需要针对性配置：
| 周期 | 信号频率 | 噪声水平 | 推荐核心指标 | 推荐可选指标 | 止损倍数(ATR) |
|------|----------|----------|-------------|-------------|--------------|
| 1m~3m | 极高 | 极高 | EWO + MACD | — | 1.5x |
| 5m~10m | 高 | 高 | EWO + MACD + AO | KDJ | 2.0x |
| 15m~30m | 中高 | 中 | EWO + MACD + AO + MA | KDJ + RSI | 2.5x |
| 1h | 中 | 中低 | 全核心 | KDJ + RSI + SuperTrend | 3.0x |
| 4h | 低 | 低 | 全核心 | 全部可选 | 3.5x |
| 1d | 极低 | 极低 | 全核心 | 全部可选 + DMI | 4.0x |
### 3.2 加密货币主流币专项配置
针对 BTC、ETH、SOL 等主流币的特性，建议以下专项配置：
**BTC（10m 周期）推荐配置**：
- EWO 强力穿越阈值：`|EWO| > 15`
- MACD 参数：维持 (10, 20, 10)，与 tradehk 一致
- 大周期过滤：启用 4h 偏向，粘性确认次数 3
- 额外过滤：CME 交易时段权重加成（09:30~16:00 ET）
- 信号强度要求：10m 周期至少 MODERATE
**ETH（10m 周期）推荐配置**：
- EWO 强力穿越阈值：`|EWO| > 5`
- 额外指标：启用 SuperTrend（ATR 10, 乘数 3）
- Gas 费用异常检测：Gas 暴涨时降低信号权重
**SOL（10m 周期）推荐配置**：
- EWO 强力穿越阈值：`|EWO| > 0.5`（防止微弱穿越误触发）
- 额外过滤：Stoch RSI 确认（K > 20 才允许做多信号）
- 注意：SOL 的 EWO 绝对值天然较小，必须使用相对幅度判断
**XAUT（黄金代币，4h 周期）推荐配置**：
- EWO 强力穿越阈值：`|EWO| > 3`
- 额外指标：启用 MFI（资金流量指数），黄金市场受机构资金影响大
- 大周期过滤：使用 12h 偏向
- 注意：黄金波动率低（年化 12-18%），信号频率天然低，不宜强求
### 3.3 多周期联动信号架构（MTF 框架）
建议实现三层周期联动，以 10m 交易为例：
```
确认层（4h）：assessBigTimeframeBias → BULLISH/BEARISH/NEUTRAL
↓ 仅在 BULLISH 时允许做多信号，BEARISH 时允许做空信号
过滤层（1h）：EWO 方向 + MACD 方向 → 中周期偏向确认
↓ 中周期与大周期方向一致时，信号权重加成 +1
执行层（10m）：完整信号引擎 → 触发实际交易信号
```
实现思路：
```typescript
// 建议：三层 MTF 信号架构
interface MTFSignalConfig {
confirmInterval: '4h' | '12h'; // 确认层周期
filterInterval: '1h' | '4h'; // 过滤层周期
executeInterval: TimeInterval; // 执行层周期
requireAllAligned: boolean; // 是否要求三层全部对齐
}
function generateMTFSignal(
confirmCandles: Candle[],
filterCandles: Candle[],
executeCandles: Candle[],
config: MTFSignalConfig,
params: IndicatorParams
): TradingSignal | null {
// 1. 大周期确认
const bigBias = assessBigTimeframeBias(confirmCandles);
if (bigBias === 'NEUTRAL') return null;
// 2. 中周期过滤（简化版 EWO + MACD 方向）
const midIndicators = calculateAllIndicators(filterCandles);
const midLast = filterCandles.length - 1;
const midEwo = midIndicators.ewo[midLast];
const midMidBias = midEwo > 0 ? 'BULLISH' : 'BEARISH';
// 3. 方向一致性检查
if (config.requireAllAligned && midMidBias !== bigBias) return null;
// 4. 执行层信号生成（带方向过滤）
const signal = generateSignal(executeCandles, symbol, config.executeInterval, params);
if (!signal) return null;
// 5. 方向过滤：只允许与大周期一致的信号
if (bigBias === 'BULLISH' && signal.type === 'SELL') return null;
if (bigBias === 'BEARISH' && signal.type === 'BUY') return null;
// 6. 中周期对齐时，提升信号强度
if (midMidBias === bigBias && signal.strength === 'MODERATE') {
signal.strength = 'STRONG';
signal.reasons.push('✅ 多周期方向对齐，信号强度提升');
}
return signal;
}
```
---
## 四、EWO 通知系统优化建议
### 4.1 通知内容增强
现有通知已包含关键信息（转换时间、EWO 值、持续时间），建议增加以下字段：
```
EWO转换提醒 BTC/10m ← 现有
EWO 红 -> 绿 ← 现有
EWO: -29.05 -> +33.32 ← 现有
上一阶段持续: 27根K线 ← 现有
【建议新增】
EWO 穿越强度: 强（绝对值 33.32 > 阈值 15.0）
MACD 方向: 金叉确认 ✅ / 未确认 ❌
4h 大周期偏向: BULLISH ✅ / BEARISH ❌ / NEUTRAL ⚠️
成交量: 放量 ✅ / 缩量 ⚠️ / 正常
建议操作: 等待 MACD 金叉确认后做多 / 谨慎（大周期未对齐）
```
### 4.2 EWO 转换规则分级
建议将 `EwoTurnAlertRule` 扩展为分级触发：
```typescript
// 建议扩展 EwoTurnAlertRule
export interface EwoTurnAlertRuleV2 extends EwoTurnAlertRule {
// 新增字段
minEwoAbsValue: number; // EWO 穿越后的最小绝对值（防假突破）
requireMacdConfirm: boolean; // 是否要求 MACD 同向确认
requireVolumeExpansion: boolean; // 是否要求放量确认
minPhaseDuration: number; // 上一阶段最少持续 K 线数
notifyLevel: 'all' | 'strong_only'; // 通知级别
}
```
---
## 五、未来调整路线图
### 5.1 短期优化（1-2 周内可实现）
1. **EWO 幅度过滤**：在 `generateSignal` 中加入 `minEwoAbsValue` 参数，按品种配置阈值。优先级最高，可立即消除 SOL 类假信号。
2. **周期自适应阈值**：修改 `strongThreshold` 计算，引入 `intervalMultiplier`。
3. **EWO 通知增强**：在飞书通知中增加 MACD 确认状态和大周期偏向字段。
### 5.2 中期优化（1 个月内）
4. **EWO 阶段持续时间奖励**：实现 `countEwoPhaseDuration`，在穿越时加入持续时间奖励分。
5. **放量确认机制**：实现 `detectVolumeExpansion`，放量时信号加权。
6. **粘性大周期偏向**：实现 `StickyBiasState`，防止震荡期频繁切换。
7. **XAUT 专项配置**：针对黄金代币的低波动特性，优化 MFI 权重和 EWO 阈值。
### 5.3 长期优化（3 个月内）
8. **MTF 三层信号架构**：实现完整的多周期联动信号系统，支持 4h→1h→10m 三层过滤。
9. **品种自适应参数**：基于历史回测自动优化各品种的 EWO 阈值、MACD 参数。
10. **AI 辅助信号过滤**：使用 LightGBM 或简单神经网络，基于历史信号质量数据训练过滤模型，自动识别高质量信号。
11. **链上数据融合**：将 Glassnode 的 SOPR、交易所净流入等链上指标融入信号评分，提升 BTC/ETH 信号准确率。
---
## 六、优化效果预期
基于历史回测数据的理论估算（需实际验证）：
| 优化项 | 预期假信号减少 | 预期胜率提升 | 实现难度 |
|--------|--------------|-------------|---------|
| EWO 幅度过滤 | 30-40% | +3-5% | 低 |
| 周期自适应阈值 | 15-25% | +2-3% | 低 |
| EWO 持续时间奖励 | 10-15% | +1-2% | 低 |
| 放量确认机制 | 20-30% | +3-4% | 低 |
| 粘性大周期偏向 | 10-20% | +1-2% | 中 |
| MTF 三层架构 | 40-50% | +5-8% | 高 |
| AI 辅助过滤 | 50-60% | +8-12% | 高 |
> **注意**：以上数据为理论估算，实际效果需通过严格回测验证。不同市场状态下效果差异显著。
---
## 七、参考资源
- tradehk 源码：`client/src/lib/indicators.ts`（信号引擎核心）
- tradehk 源码：`client/src/lib/types.ts`（类型定义，含 `EwoTurnAlertRule`）
- arXiv 2511.00665：Technical Analysis Meets Machine Learning: Bitcoin Evidence
- arXiv 2503.21422：From Deep Learning to LLMs: A survey of AI in Quantitative Investment
- Freqtrade 文档：https://www.freqtrade.io/en/stable/strategy-customization/
- Pine Script 多周期函数：https://www.tradingview.com/pine-script-docs/concepts/timeframes/

552
samples/optimized_signal_engine.py 普通文件
查看文件

@@ -0,0 +1,552 @@
"""
优化版信号引擎 - 基于 tradehk 源码分析的改进实现
主要优化点：
1. EWO 幅度过滤（防止 SOL 类微弱穿越假信号）
2. 周期自适应信号强度阈值
3. EWO 阶段持续时间奖励
4. 放量确认加权
5. 粘性大周期偏向过滤
6. MTF 三层周期联动框架
使用方法：
pip install pandas numpy requests
python optimized_signal_engine.py
"""
import pandas as pd
import numpy as np
import requests
from dataclasses import dataclass, field
from typing import Optional, Literal
from datetime import datetime
# ============================================================
# 类型定义
# ============================================================
TimeInterval = Literal['1m','3m','5m','10m','15m','30m','1h','4h','12h','1d','1w']
TrendBias = Literal['BULLISH','BEARISH','NEUTRAL']
SignalType = Literal['BUY','SELL','NEUTRAL']
SignalStrength = Literal['STRONG','MODERATE','WEAK']
# EWO 幅度阈值（按品种配置，防止微弱穿越假信号）
EWO_STRONG_THRESHOLD: dict[str, float] = {
'BTCUSDT': 15.0,
'ETHUSDT': 5.0,
'SOLUSDT': 0.5,
'BNBUSDT': 2.0,
'DOGEUSDT': 0.001,
'XAUTUSDT': 3.0,
'DEFAULT': 1.0,
}
# 周期自适应阈值乘数（短周期需要更高评分才触发强信号）
INTERVAL_MULTIPLIER: dict[str, float] = {
'1m': 1.5, '3m': 1.4, '5m': 1.3, '10m': 1.2,
'15m': 1.1, '30m': 1.0, '1h': 0.9, '4h': 0.8,
'12h': 0.7, '1d': 0.6, '1w': 0.5,
}
@dataclass
class SignalResult:
signal_type: SignalType
strength: SignalStrength
bullish_score: int
bearish_score: int
reasons: list[str]
price: float
timestamp: str
ewo_value: float
ewo_strength: str # 'STRONG' / 'WEAK' / 'NONE'
@dataclass
class StickyBiasState:
"""粘性大周期偏向状态（防止震荡期频繁切换）"""
bias: TrendBias = 'NEUTRAL'
consecutive_confirms: int = 0
min_confirms: int = 3 # 需要连续 N 次确认才切换方向
# ============================================================
# 数据获取
# ============================================================
def fetch_klines(symbol: str, interval: str, limit: int = 500) -> pd.DataFrame:
url = "https://api.binance.com/api/v3/klines"
resp = requests.get(url, params={"symbol": symbol, "interval": interval, "limit": limit}, timeout=10)
resp.raise_for_status()
df = pd.DataFrame(resp.json(), columns=[
'timestamp','open','high','low','close','volume',
'close_time','quote_volume','trades','taker_buy_base','taker_buy_quote','ignore'
])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
for col in ['open','high','low','close','volume']:
df[col] = df[col].astype(float)
return df[['timestamp','open','high','low','close','volume']].set_index('timestamp')
# ============================================================
# 指标计算
# ============================================================
def ema(s: pd.Series, n: int) -> pd.Series:
return s.ewm(span=n, adjust=False).mean()
def sma(s: pd.Series, n: int) -> pd.Series:
return s.rolling(n).mean()
def rma(s: pd.Series, n: int) -> pd.Series:
return s.ewm(alpha=1/n, adjust=False).mean()
def calc_ewo(close: pd.Series) -> pd.Series:
"""EWO = EMA(5) - EMA(35)"""
return ema(close, 5) - ema(close, 35)
def calc_macd(close: pd.Series, fast=10, slow=20, signal=10):
macd_line = ema(close, fast) - ema(close, slow)
signal_line = ema(macd_line, signal)
return macd_line, signal_line, macd_line - signal_line
def calc_ao(df: pd.DataFrame) -> pd.Series:
mid = (df['high'] + df['low']) / 2
return sma(mid, 5) - sma(mid, 34)
def calc_rsi(close: pd.Series, period=14) -> pd.Series:
delta = close.diff()
gain = rma(delta.clip(lower=0), period)
loss = rma((-delta).clip(lower=0), period)
rs = gain / loss.replace(0, np.nan)
return 100 - 100 / (1 + rs)
def calc_atr(df: pd.DataFrame, period=14) -> pd.Series:
tr = pd.concat([
df['high'] - df['low'],
(df['high'] - df['close'].shift(1)).abs(),
(df['low'] - df['close'].shift(1)).abs()
], axis=1).max(axis=1)
return rma(tr, period)
def calc_kdj(df: pd.DataFrame, period=9):
low_min = df['low'].rolling(period).min()
high_max = df['high'].rolling(period).max()
rsv = (df['close'] - low_min) / (high_max - low_min).replace(0, np.nan) * 100
k = rsv.ewm(com=2, adjust=False).mean()
d = k.ewm(com=2, adjust=False).mean()
j = 3 * k - 2 * d
return k, d, j
def calc_supertrend(df: pd.DataFrame, period=10, multiplier=3.0):
atr = calc_atr(df, period)
hl2 = (df['high'] + df['low']) / 2
upper = hl2 + multiplier * atr
lower = hl2 - multiplier * atr
direction = pd.Series(1, index=df.index)
for i in range(1, len(df)):
if upper.iloc[i] > upper.iloc[i-1] and df['close'].iloc[i-1] <= upper.iloc[i-1]:
upper.iloc[i] = upper.iloc[i-1]
if lower.iloc[i] < lower.iloc[i-1] and df['close'].iloc[i-1] >= lower.iloc[i-1]:
lower.iloc[i] = lower.iloc[i-1]
if direction.iloc[i-1] == 1:
direction.iloc[i] = -1 if df['close'].iloc[i] < lower.iloc[i] else 1
else:
direction.iloc[i] = 1 if df['close'].iloc[i] > upper.iloc[i] else -1
return direction
# ============================================================
# 优化一：EWO 幅度过滤
# ============================================================
def get_ewo_threshold(symbol: str) -> float:
"""获取品种对应的 EWO 强力穿越阈值"""
return EWO_STRONG_THRESHOLD.get(symbol.upper(), EWO_STRONG_THRESHOLD['DEFAULT'])
def classify_ewo_crossing(ewo_now: float, ewo_prev: float, symbol: str) -> tuple[str, int, str]:
"""
EWO 穿越分类（优化版）
返回：(穿越类型, 评分, 说明)
"""
threshold = get_ewo_threshold(symbol)
if ewo_now > 0 and ewo_prev <= 0:
# 上穿零轴
if abs(ewo_now) >= threshold:
return 'STRONG_CROSS_UP', 2, f'EWO 强力上穿零轴 (值={ewo_now:.4f} ≥ 阈值{threshold})'
else:
return 'WEAK_CROSS_UP', 1, f'EWO 微弱上穿零轴 (值={ewo_now:.4f} < 阈值{threshold}) ⚠️ 注意假突破'
elif ewo_now < 0 and ewo_prev >= 0:
# 下穿零轴
if abs(ewo_now) >= threshold:
return 'STRONG_CROSS_DOWN', 2, f'EWO 强力下穿零轴 (值={ewo_now:.4f})'
else:
return 'WEAK_CROSS_DOWN', 1, f'EWO 微弱下穿零轴 (值={ewo_now:.4f}) ⚠️ 注意假突破'
elif ewo_now > 0:
return 'ABOVE_ZERO', 1, f'EWO 在零轴上方 (值={ewo_now:.4f})'
elif ewo_now < 0:
return 'BELOW_ZERO', 1, f'EWO 在零轴下方 (值={ewo_now:.4f})'
return 'NEUTRAL', 0, 'EWO 在零轴附近'
# ============================================================
# 优化二：EWO 阶段持续时间计算
# ============================================================
def count_ewo_phase_duration(ewo_series: pd.Series, current_idx: int) -> int:
"""计算当前 EWO 阶段（正/负）在穿越前持续了多少根 K 线"""
if current_idx <= 0:
return 0
prev_sign = 1 if ewo_series.iloc[current_idx - 1] >= 0 else -1
count = 0
for i in range(current_idx - 1, -1, -1):
sign = 1 if ewo_series.iloc[i] >= 0 else -1
if sign == prev_sign:
count += 1
else:
break
return count
# ============================================================
# 优化三：放量/缩量检测
# ============================================================
def detect_volume_expansion(volumes: pd.Series, lookback=5, threshold=1.5) -> bool:
"""检测放量（最近 lookback 根 K 线均量 > 前 lookback 根均量 * threshold）"""
if len(volumes) < lookback * 2:
return False
recent_avg = volumes.iloc[-lookback:].mean()
prev_avg = volumes.iloc[-lookback*2:-lookback].mean()
return recent_avg > prev_avg * threshold if prev_avg > 0 else False
def detect_volume_contraction(volumes: pd.Series, lookback=5, threshold=0.7) -> bool:
"""检测缩量"""
if len(volumes) < lookback * 2:
return False
recent_avg = volumes.iloc[-lookback:].mean()
prev_avg = volumes.iloc[-lookback*2:-lookback].mean()
return recent_avg < prev_avg * threshold if prev_avg > 0 else False
# ============================================================
# 优化四：大周期偏向评估（粘性版）
# ============================================================
def assess_big_timeframe_bias(df: pd.DataFrame) -> TrendBias:
"""评估大周期偏向（EWO+MACD+AO 综合评分）"""
if len(df) < 35:
return 'NEUTRAL'
close = df['close']
ewo = calc_ewo(close)
macd_line, signal_line, histogram = calc_macd(close)
ao = calc_ao(df)
last = -1
bull, bear = 0, 0
ewo_val = ewo.iloc[last]
if not np.isnan(ewo_val):
if ewo_val > 0: bull += 2
else: bear += 2
macd_val = macd_line.iloc[last]
sig_val = signal_line.iloc[last]
if not np.isnan(macd_val) and not np.isnan(sig_val):
if macd_val > sig_val: bull += 1
else: bear += 1
hist_val = histogram.iloc[last]
if not np.isnan(hist_val):
if hist_val > 0: bull += 1
else: bear += 1
ao_val = ao.iloc[last]
if not np.isnan(ao_val):
if ao_val > 0: bull += 1
else: bear += 1
if bear >= 4: return 'BEARISH'
if bull >= 4: return 'BULLISH'
return 'NEUTRAL'
def update_sticky_bias(state: StickyBiasState, new_bias: TrendBias) -> StickyBiasState:
"""更新粘性偏向状态"""
if new_bias == state.bias:
state.consecutive_confirms += 1
return state
if state.consecutive_confirms >= state.min_confirms:
return StickyBiasState(bias=new_bias, consecutive_confirms=1, min_confirms=state.min_confirms)
# 尚未达到切换阈值，保持原方向
return state
# ============================================================
# 主信号引擎（优化版）
# ============================================================
def generate_optimized_signal(
df: pd.DataFrame,
symbol: str,
interval: str,
big_bias: TrendBias = 'NEUTRAL',
use_kdj: bool = True,
use_supertrend: bool = True,
use_rsi: bool = True,
) -> Optional[SignalResult]:
"""
优化版多指标共振信号引擎
改进点：
1. EWO 幅度过滤（按品种阈值分级评分）
2. EWO 阶段持续时间奖励
3. 周期自适应信号强度阈值
4. 放量确认加权
5. 大周期方向过滤
"""
if len(df) < 100:
return None
close = df['close']
ewo = calc_ewo(close)
macd_line, signal_line, histogram = calc_macd(close)
ao = calc_ao(df)
ma10 = sma(close, 10)
ma100 = sma(close, 100)
curr_idx = len(df) - 1
prev_idx = len(df) - 2
bull, bear = 0, 0
reasons = []
# ── 核心信号 1：EWO（优化版：幅度分级）──
ewo_now = ewo.iloc[curr_idx]
ewo_prev = ewo.iloc[prev_idx]
ewo_cross_type, ewo_score, ewo_reason = classify_ewo_crossing(ewo_now, ewo_prev, symbol)
if 'UP' in ewo_cross_type:
bull += ewo_score
reasons.append(ewo_reason)
# 持续时间奖励
prev_phase_len = count_ewo_phase_duration(ewo, curr_idx)
if prev_phase_len >= 20:
bull += 1
reasons.append(f' ↳ 前空头阶段持续 {prev_phase_len} 根K线，反转可信度高 (+1)')
elif 'DOWN' in ewo_cross_type:
bear += ewo_score
reasons.append(ewo_reason)
prev_phase_len = count_ewo_phase_duration(ewo, curr_idx)
if prev_phase_len >= 20:
bear += 1
reasons.append(f' ↳ 前多头阶段持续 {prev_phase_len} 根K线，反转可信度高 (+1)')
elif ewo_score > 0:
if 'ABOVE' in ewo_cross_type:
bull += ewo_score
else:
bear += ewo_score
reasons.append(ewo_reason)
ewo_strength_label = 'STRONG' if 'STRONG' in ewo_cross_type else ('WEAK' if 'WEAK' in ewo_cross_type else 'NONE')
# ── 核心信号 2：MACD ──
macd_now = macd_line.iloc[curr_idx]
macd_prev = macd_line.iloc[prev_idx]
sig_now = signal_line.iloc[curr_idx]
sig_prev = signal_line.iloc[prev_idx]
hist_now = histogram.iloc[curr_idx]
hist_prev = histogram.iloc[prev_idx]
if macd_now > sig_now and macd_prev <= sig_prev:
bull += 2; reasons.append('MACD 金叉 (+2)')
elif macd_now < sig_now and macd_prev >= sig_prev:
bear += 2; reasons.append('MACD 死叉 (+2)')
if not np.isnan(hist_now) and not np.isnan(hist_prev):
if hist_now > 0 and hist_now > hist_prev:
bull += 1; reasons.append('MACD 柱状图正向扩大 (+1)')
elif hist_now < 0 and hist_now < hist_prev:
bear += 1; reasons.append('MACD 柱状图负向扩大 (+1)')
# ── 核心信号 3：AO ──
ao_now = ao.iloc[curr_idx]
ao_prev = ao.iloc[prev_idx]
if ao_now > 0 and ao_prev <= 0:
bull += 1; reasons.append('AO 上穿零轴 (+1)')
elif ao_now < 0 and ao_prev >= 0:
bear += 1; reasons.append('AO 下穿零轴 (+1)')
# ── 核心信号 4：MA 排列 ──
ma10_now = ma10.iloc[curr_idx]
ma100_now = ma100.iloc[curr_idx]
price_now = close.iloc[curr_idx]
if not np.isnan(ma10_now) and not np.isnan(ma100_now):
if price_now > ma10_now > ma100_now:
bull += 1; reasons.append('多头排列 价格>MA10>MA100 (+1)')
elif price_now < ma10_now < ma100_now:
bear += 1; reasons.append('空头排列 价格<MA10<MA100 (+1)')
# ── 可选信号：RSI ──
if use_rsi:
rsi = calc_rsi(close)
rsi_now = rsi.iloc[curr_idx]
rsi_prev = rsi.iloc[prev_idx]
if not np.isnan(rsi_now):
if rsi_now < 30:
bull += 1; reasons.append(f'RSI 超卖 {rsi_now:.1f} (+1)')
elif rsi_now > 70:
bear += 1; reasons.append(f'RSI 超买 {rsi_now:.1f} (+1)')
if not np.isnan(rsi_prev):
if rsi_prev < 30 <= rsi_now:
bull += 1; reasons.append(f'RSI 从超卖区回升 (+1)')
elif rsi_prev > 70 >= rsi_now:
bear += 1; reasons.append(f'RSI 从超买区回落 (+1)')
# ── 可选信号：KDJ ──
if use_kdj:
k, d, j = calc_kdj(df)
k_now, d_now = k.iloc[curr_idx], d.iloc[curr_idx]
k_prev, d_prev = k.iloc[prev_idx], d.iloc[prev_idx]
if not any(np.isnan([k_now, d_now, k_prev, d_prev])):
if k_prev <= d_prev and k_now > d_now:
if k_now < 30:
bull += 2; reasons.append(f'KDJ 低位金叉 K={k_now:.1f} (+2)')
else:
bull += 1; reasons.append(f'KDJ 金叉 K={k_now:.1f} (+1)')
elif k_prev >= d_prev and k_now < d_now:
if k_now > 70:
bear += 2; reasons.append(f'KDJ 高位死叉 K={k_now:.1f} (+2)')
else:
bear += 1; reasons.append(f'KDJ 死叉 K={k_now:.1f} (+1)')
# ── 可选信号：SuperTrend ──
if use_supertrend:
st_dir = calc_supertrend(df)
dir_now = st_dir.iloc[curr_idx]
dir_prev = st_dir.iloc[prev_idx]
if dir_prev == -1 and dir_now == 1:
bull += 2; reasons.append('SuperTrend 转多 (+2)')
elif dir_prev == 1 and dir_now == -1:
bear += 2; reasons.append('SuperTrend 转空 (+2)')
elif dir_now == 1:
bull += 1; reasons.append('SuperTrend 多头趋势中 (+1)')
elif dir_now == -1:
bear += 1; reasons.append('SuperTrend 空头趋势中 (+1)')
# ── 优化：放量确认 ──
if detect_volume_expansion(df['volume']):
if bull > bear:
bull += 1; reasons.append('✅ 放量确认多头信号 (+1)')
elif bear > bull:
bear += 1; reasons.append('✅ 放量确认空头信号 (+1)')
elif detect_volume_contraction(df['volume']):
reasons.append('⚠️ 缩量中，信号可靠性降低')
# ── 大周期偏向过滤 ──
if big_bias != 'NEUTRAL':
if big_bias == 'BULLISH' and bear > bull:
reasons.append(f'⚠️ 大周期偏向 {big_bias}，空头信号被过滤')
return None
elif big_bias == 'BEARISH' and bull > bear:
reasons.append(f'⚠️ 大周期偏向 {big_bias}，多头信号被过滤')
return None
# 大周期对齐时加权
if (big_bias == 'BULLISH' and bull > bear) or (big_bias == 'BEARISH' and bear > bull):
if bull > bear: bull += 1
else: bear += 1
reasons.append(f'✅ 大周期 {big_bias} 对齐，信号加权 (+1)')
# ── 周期自适应强度判定 ──
mult = INTERVAL_MULTIPLIER.get(interval, 1.0)
active_optional = sum([use_rsi, use_kdj, use_supertrend])
strong_threshold = int(np.ceil((5 + active_optional * 0.5) * mult))
moderate_threshold = int(np.ceil((3 + active_optional * 0.3) * mult))
if bull == 0 and bear == 0:
return None
if bull > bear:
sig_type: SignalType = 'BUY'
score = bull
elif bear > bull:
sig_type = 'SELL'
score = bear
else:
return None
if score >= strong_threshold:
strength: SignalStrength = 'STRONG'
elif score >= moderate_threshold:
strength = 'MODERATE'
else:
strength = 'WEAK'
return SignalResult(
signal_type=sig_type,
strength=strength,
bullish_score=bull,
bearish_score=bear,
reasons=reasons,
price=price_now,
timestamp=df.index[-1].strftime('%Y-%m-%d %H:%M:%S'),
ewo_value=ewo_now,
ewo_strength=ewo_strength_label,
)
# ============================================================
# 主程序：对比原版 vs 优化版
# ============================================================
if __name__ == '__main__':
print("=" * 65)
print("优化版信号引擎 - 基于 tradehk 源码改进")
print("=" * 65)
test_cases = [
('BTCUSDT', '10m', 'BTC/10m'),
('SOLUSDT', '10m', 'SOL/10m'),
('ETHUSDT', '4h', 'ETH/4h'),
('XAUTUSDT', '4h', 'XAUT/4h（黄金代币）'),
]
for symbol, interval, label in test_cases:
print(f"\n{'─' * 55}")
print(f"品种：{label}")
print(f"{'─' * 55}")
try:
df = fetch_klines(symbol, interval, limit=300)
# 获取大周期偏向（4h）
df_4h = fetch_klines(symbol, '4h', limit=100)
big_bias = assess_big_timeframe_bias(df_4h)
print(f"大周期偏向（4h）：{big_bias}")
# 运行优化版信号引擎
result = generate_optimized_signal(
df, symbol, interval,
big_bias=big_bias,
use_kdj=True,
use_supertrend=True,
use_rsi=True,
)
if result:
emoji = '🟢' if result.signal_type == 'BUY' else '🔴'
print(f"\n{emoji} 信号：{result.signal_type} ({result.strength})")
print(f" 价格：{result.price:.4f}")
print(f" EWO：{result.ewo_value:.4f} [{result.ewo_strength}]")
print(f" 多头评分：{result.bullish_score} 空头评分：{result.bearish_score}")
print(f"\n 触发原因：")
for r in result.reasons:
print(f" • {r}")
else:
print(" 无信号（被过滤或评分不足）")
except Exception as e:
print(f" 错误：{e}")
print(f"\n{'=' * 65}")
print("优化要点说明：")
print(" 1. EWO 幅度过滤：SOL 需 |EWO| > 0.5 才触发强信号（防假突破）")
print(" 2. 周期自适应：10m 周期阈值 × 1.2，需更高评分才触发 STRONG")
print(" 3. 持续时间奖励：前阶段 ≥ 20 根K线时，反转信号额外 +1")
print(" 4. 放量确认：放量时信号加权 +1")
print(" 5. 大周期过滤：与 4h 偏向相反的信号被过滤")
print("=" * 65)
print("⚠️ 仅供学习研究，不构成投资建议")