mike/Hermes-Skills
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(strategy-management): exit_levels.py - 港美股做T 出场点位算法 (混合公式)

Browse Source 
方法 2 (百分比波动率) + 方法 3 (关键价位) 混合算法

feat(intraday-regime-detector): 新 skill - 日内市场状态判别

来源: DeepSeek chat share 26iikphv8h94feze9q
核心: R² + ADF + 历史波动率, 识别趋势市 / 震荡市 / 混乱
推荐: 趋势跟踪 / 网格交易 / 布林带回归 / NO_TRADE
This commit is contained in:
mike
2026-07-18 21:26:14 +08:00
parent 533c342305
commit e23f8d38c0
5 changed files with 1133 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
intraday-regime-detector/SKILL.md
+204
View File
@@ -0,0 +1,204 @@
---
name: intraday-regime-detector
description: "港美股日内做T 元策略 - 根据 5min K 线自动判别市场状态 (趋势/震荡/混乱) 并推荐匹配策略 (趋势跟踪/网格/布林带回归)。来源 DeepSeek 分享, 决策树: R² > 0.75 趋势; R² < 0.30 + ADF 平稳 + 低波动 = 网格; 高波动 = 布林带; 其他 NO_TRADE。⚠️ 仅识别市场状态, 不替代 longbridge-t-monitor / strategy-management 的入场/出场逻辑。"
version: 1.0.0
author: Hermes Agent + DeepSeek 分享 (26iikphv8h94feze9q)
tags: [trading, intraday, market-regime, regime-detection, deepseek, hk, us]
metadata:
hermes:
tags: [trading, intraday, market-regime, regime-detection, deepseek, hk, us]
related_skills: [longbridge-t-monitor, strategy-management]
scripts:
- intraday_regime.py: "核心: IntradayStrategySelector + MarketDiagnosis + MarketRegime + StrategyType"
- regime_scan.py: "扫描港美股 top 5 候选, 拉长桥 5min K线, 跑判别, 推报告"
references:
- decision-tree.md: "决策树详细说明 + 4 策略参数说明"
---
# Intraday Regime Detector (港美股日内做T 元策略)
**核心定位**:**不是替代** `longbridge-t-monitor` 或 `strategy-management`, 而是**在它们之前**先判断"现在适不适合做T、做哪个策略"。
```
intraday-regime-detector (本 skill) → 告诉用户 "用什么策略 + 为什么"
↓
longbridge-t-monitor (现有) → 执行入场/出场
strategy-management (现有) → 选策略 + 算 SL/TP
```
## 🎯 解决的痛点
| 痛点 | 解决 |
|---|---|
| 趋势市用网格 = 反复止损 | R² > 0.75 → 强制趋势策略 |
| 震荡市用趋势 = 追涨杀跌 | R² < 0.30 → 强制震荡策略 |
| 混乱行情硬做 = 越做越亏 | ADF p > 0.05 → NO_TRADE |
| 不知道用宽网格还是窄网格 | 波动率高 → 布林带, 低 → 网格 |
## 📦 决策树 (DeepSeek 原始版)
```
第一步: 输入近 20-30 根 5 分钟 K 线 + 开盘缺口
第二步: 计算趋势效率 R² (线性回归)
├─ R² > 0.75: 强趋势市 → 趋势跟踪 (顺势 EMA5 支撑/阻力)
├─ R² < 0.30: 强震荡市
│ ├─ ADF p < 0.05 (平稳):
│ │ ├─ 波动率 > 30%: 布林带回归
│ │ └─ 波动率 ≤ 30%: 网格交易
│ └─ ADF p ≥ 0.05 (不平稳): 混乱 → NO_TRADE
└─ 0.30 ≤ R² ≤ 0.75: 过渡 → 暂停, 等模式清晰
```
## 🚀 快速使用
### Python API
```python
import sys
sys.path.insert(0, '/home/openclaw/.hermes/skills/trading/intraday-regime-detector/scripts')
from intraday_regime import IntradayStrategySelector
# 假设 df 是 5min K线 DataFrame, 包含 open/high/low/close
selector = IntradayStrategySelector()
diagnosis = selector.diagnose(df, open_gap_pct=0.3)
print(f"状态: {diagnosis.regime.value}")
print(f"推荐: {diagnosis.recommended_strategy.value}")
print(f"R²: {diagnosis.r_squared}, 置信度: {diagnosis.confidence}")
print(f"参数: {diagnosis.strategy_params}")
```
### CLI 扫描 (港美股 top 5)
```bash
/home/openclaw/.hermes/hermes-agent/venv/bin/python \
/home/openclaw/.hermes/skills/trading/intraday-regime-detector/scripts/regime_scan.py
```
输出示例:
```
🔲 9888.HK (HK) 现价 $110.30 (+1.21%) 置信度 85%
状态: 低波震荡 | R²=0.2363 | 波动率=3.32% | ADF p=0.01
推荐: 网格交易做T
grid_spacing: 0.500%
grid_levels: 3
base_price: 110.30
```
## 📊 输出数据结构
`MarketDiagnosis` (dataclass):
```python
@dataclass
class MarketDiagnosis:
regime: MarketRegime # 6 种状态之一
r_squared: float # 趋势效率 0-1
volatility: float # 年化波动率
adf_pvalue: float # ADF 平稳检验 p 值
recommended_strategy: StrategyType # 4 种策略之一
strategy_params: Dict # 动态参数 (SL/TP/grid 等)
confidence: float # 0-1
reasoning: str # 人话解释
```
`MarketRegime` 枚举:
- `STRONG_TREND_UP` / `STRONG_TREND_DOWN`
- `HIGH_VOL_SHAKE` (高波动震荡)
- `LOW_VOL_STABLE` (低波动震荡)
- `CHAOTIC` (混乱)
- `UNKNOWN` (过渡区间)
`StrategyType` 枚举:
- `TREND_FOLLOWING` (EMA5 顺势)
- `GRID_TRADING` (3 格 × 0.5%)
- `BOLLINGER_REVERSION` (20 期 ±2σ)
- `NO_TRADE` (暂停)
## 🔧 配置参数
```python
selector = IntradayStrategySelector(
trend_r2_threshold=0.75, # R² 高于此 = 趋势市
chaos_r2_threshold=0.30, # R² 低于此 = 震荡市
adf_significance=0.05, # ADF p < 此 = 平稳
vol_lookback=20, # 历史波动率窗口
high_vol_threshold=0.30, # 年化波动率高/低分界
grid_count=3, # 网格层数
bb_period=20, # 布林带周期
bb_std=2.0, # 布林带 σ
ema_period=5, # 趋势策略 EMA 周期
)
```
## 🔄 与现有 skill 的关系
| Skill | 关系 |
|---|---|
| `longbridge-t-monitor` | **下游** - 本 skill 决定"该不该做T + 用什么策略", 然后 longbridge-t-monitor 执行 |
| `strategy-management` | **互补** - strategy-management 有具体的 5 策略 (rsi2_revert/vwap_revert/early_bird/turtle/sma), 本 skill 是"先用元策略筛一下再用具体策略" |
| `intraday-trading` | **理论来源** - 已有 4 策略设计 + 5 步预检 + 资金管理表 |
| `crypto-t-monitor` | **独立** - 币圈用 OKX ATR 公式, 本 skill 不涉及 |
**集成路径 (建议)**:
```
盘前 cron (c3401d727f39, cfa0c1d6baa5)
↓ 生成候选池
盘中 cron (新加): regime_scan
↓ 输出 "可做 T 的票 + 推荐策略"
手动 / agent: 看推送
↓ 决定是否入场
longbridge-t-monitor: 执行
```
## ⚠️ Pitfalls
1. **ADF 是简化版** — 实际生产用 `statsmodels.tsa.stattools.adfuller`. 代码已 fallback, 有 statsmodels 就用真 ADF
2. **30 根 K 线窗口** — 跟 longbridge-t-monitor 一样的限制, period 选 5m → 2.5h 窗口
3. **网格/布林带参数是建议值** — 实盘要按资金 + 流动性 + 个人风险偏好调整
4. **本 skill 不替代风控** — 出场点位 / 仓位管理 / 单日最大亏损 → 用 longbridge-t-monitor
5. **不主动下单 (用户偏好 2026-07-16)**: 用户原话 "在跑日内交易扫描任务时使用 intraday-regime-detector, 不主动下单". 本 skill 只输出状态 + 推荐策略, **不要在 diagnose() 里加任何下单逻辑**.
6. **港股 candlesticks 不支持 `--json`**: 长桥 CLI 港股 candlesticks 只输出中文表格, 表格分隔符是 `│` (不是 `|`). `regime_scan.py` 已处理. 美股用 `--json`.
## 👤 用户偏好 (2026-07-16)
- **来源**: https://chat.deepseek.com/share/26iikphv8h94feze9q
- **要求**: 跑日内交易扫描任务时使用 intraday-regime-detector, 不主动下单
- **三段式 pipeline** (已部署):
1. 候选池 (quant-factor-mining) → top 5
2. 元策略 (本 skill) → confidence ≥ 0.6 算 actionable
3. 点位 (strategy-management/exit_levels) → SL/TP/TP2
- **部署位置**:
- `/home/openclaw/qdrant/calc_hk_levels.py` - 港股
- `/home/openclaw/qdrant/calc_us_levels.py` - 美股
- Cron `c4dc9ac8854c` (港股 */15 9-15) + `70d24624637c` (美股 */15 21-3) 周一到周五
- Wrapper: `~/.hermes/scripts/hk_t_levels.sh` / `us_t_levels.sh`
- **A 股 vs 港美股** (重要区分, 用户 2026-07-16 强调):
- A 股做T = 底仓滚动 (T+1 制度)
- 港美股做T = 直接双向交易 (T+0)
- **本 skill 只服务港美股**。A 股做T 完全用不上这个。
## 🛡️ 已知问题
| 问题 | 处理 |
|---|---|
| statsmodels 未装 | 自动 fallback 到自相关近似 |
| K 线 < 10 根 | 抛 ValueError, 跳过 |
| R² 边界值 (0.30 / 0.75) | 默认参数, 可在 __init__ 调整 |
| 大量候选 NO_TRADE | 正常, 实测 10 支 7 支 NO_TRADE. 算法价值正在于"拒绝不值得做的票" |
## 📚 参考
- **来源**: https://chat.deepseek.com/share/26iikphv8h94feze9q
- **决策树详细**: `references/decision-tree.md`
- **测试数据**: `intraday_regime.py` 的 `__main__` 跑自检
- **集成代码**:
- `~/.hermes/qdrant/calc_hk_levels.py` (港股)
- `~/.hermes/qdrant/calc_us_levels.py` (美股)
- `~/.hermes/scripts/hk_t_levels.sh` / `us_t_levels.sh` (wrapper)
## 🔄 版本历史
- **v1.0.0** (2026-07-16): 初始版本
- `intraday_regime.py` - 核心判别器 (DeepSeek 原始代码 + statsmodels fallback + dataclass)
- `regime_scan.py` - 长桥 K线集成扫描器
- 自检场景 (震荡市/趋势市) 全部通过
intraday-regime-detector/references/decision-tree.md
+144
View File
@@ -0,0 +1,144 @@
# 决策树详细说明
来源: DeepSeek chat share 26iikphv8h94feze9q
## 一、核心判别算法
### 1. 趋势效率 R² (线性回归)
**目的**: 衡量趋势的"纯粹度", 比单纯看均线方向更科学。
**计算**:
- 取过去 N 根 K 线 (默认 20 根 5min K) 的收盘价序列
- 以时间 (1,2,3...20) 为自变量 X, 收盘价为因变量 Y
- 做一元线性回归
- 计算 R²
**判断**:
| R² | 状态 | 含义 |
|---|---|---|
| > 0.75 | 趋势市 | 价格运动有明确方向, 噪声小 |
| < 0.30 | 震荡市 | 价格运动无方向, 充满噪声 |
| 0.30-0.75 | 过渡 | 方向不明, 等待 |
### 2. ADF 平稳检验 (Augmented Dickey-Fuller)
**目的**: 判断价格序列是否倾向于均值回归。
**计算**:
- 对过去价格序列执行 ADF 检验
- 返回 p 值
**判断**:
| p 值 | 含义 | 策略匹配 |
|---|---|---|
| < 0.05 | 拒绝非平稳假设, 统计上平稳 | **均值回归, 适合震荡做T** |
| > 0.05 | 不能拒绝非平稳, 可能是随机游走或趋势 | **不做均值回归** |
**⚠️ 简化实现**: 本 skill 用一阶差分自相关近似, 生产建议替换为 `statsmodels.tsa.stattools.adfuller` (代码已 fallback).
### 3. 历史波动率 (年化)
**计算**: log returns 标准差 × √252
**判断**:
| 波动率 | 含义 |
|---|---|
| > 30% | 高波动, 适合宽间距逆势 (布林带) |
| ≤ 30% | 低波动, 适合网格 |
### 4. 开盘缺口
**计算**: (open - prev_close) / prev_close × 100%
**判断**:
| 缺口 | 含义 |
|---|---|
| > +0.5% | 高开强势, 优先做正T |
| < -0.5% | 低开弱势, 优先做倒T |
| 平开/微小 | 默认震荡模式 |
## 二、策略参数说明
### 1. 趋势跟踪做T (TREND_FOLLOWING)
**适用**: 强趋势市 (R² > 0.75)
**参数** (5min K):
- `direction`: long_only / short_only
- `entry_trigger`: 价格回踩 EMA5 不破 (上涨) / 价格反弹至 EMA5 受阻 (下跌)
- `stop_loss`: EMA5 × 0.995 (long) / EMA5 × 1.005 (short)
- `take_profit`: 现价 × 1.02 / × 0.98
### 2. 网格交易做T (GRID_TRADING)
**适用**: 低波动震荡 (R² < 0.30 + ADF 平稳 + 低波动)
**参数**:
- `grid_spacing`: 近期平均振幅 × 0.8, 至少 0.5%
- `grid_levels`: 3 (默认)
- `base_price`: 当前价
- `reverse_at_boundary`: True (在边界反向开仓)
### 3. 布林带回归做T (BOLLINGER_REVERSION)
**适用**: 高波动震荡 (R² < 0.30 + ADF 平稳 + 高波动)
**参数** (20 期 ±2σ):
- `upper_band`: MA20 + 2σ
- `lower_band`: MA20 - 2σ
- `sell_at_upper`: True
- `buy_at_lower`: True
- `stop_if_break`: True (破带止损)
### 4. NO_TRADE (暂停)
**适用**: 混乱或过渡状态
**参数**: `reason: 市场状态不清晰, 等待趋势或明确震荡信号`
## 三、决策流程图
```
┌─────────────────────┐
│ 输入 20 根 5min K │
│ + 开盘缺口 │
└──────────┬──────────┘
↓
┌─────────────────────┐
│ 计算 R² │
└──────────┬──────────┘
↓
┌─────────────────┼─────────────────┐
↓ ↓ ↓
R² > 0.75 0.30-0.75 R² < 0.30
强趋势 过渡 震荡
↓ ↓ ↓
TREND UNKNOWN 计算 ADF
FOLLOWING NO_TRADE ↓
┌─────┴─────┐
↓ ↓
ADF<0.05 ADF≥0.05
平稳 不平稳
↓ ↓
计算波动率 CHAOTIC
↓ NO_TRADE
┌───┴───┐
↓ ↓
vol>30% vol≤30%
↓ ↓
BOLLINGER GRID
REVERSION TRADING
```
## 四、与本 skill 的对应关系
| 步骤 | 函数 | 文件 |
|---|---|---|
| 输入校验 | `diagnose()` | intraday_regime.py |
| 计算 R² | `_calculate_r_squared()` | intraday_regime.py |
| 计算 vol | `_calculate_historical_volatility()` | intraday_regime.py |
| 计算 ADF | `_adf_test()` | intraday_regime.py |
| 趋势判断 | `_classify_regime()` | intraday_regime.py |
| 策略匹配 | `_match_strategy()` | intraday_regime.py |
| 拉 K 线 + 整合 | `regime_scan.py` | regime_scan.py |
intraday-regime-detector/scripts/intraday_regime.py
+307
View File
@@ -0,0 +1,307 @@
"""
intraday_regime.py - 日内市场状态判别 + 策略匹配
来源: DeepSeek chat share 26iikphv8h94feze9q
核心算法:
1. 趋势效率 R² (线性回归) - 判趋势 vs 震荡
2. ADF 平稳检验 - 验证均值回归
3. 历史波动率 - 区分高/低波动
4. 开盘缺口 - 识别方向偏好
决策树:
R² > 0.75 → 趋势跟踪 (顺势)
R² < 0.30 + ADF 平稳 + 低波动 → 网格交易
R² < 0.30 + ADF 平稳 + 高波动 → 布林带回归
其他 → NO_TRADE (暂停)
⚠️ 这是港美股日内做T 元策略, 跟 crypto-t-monitor / longbridge-t-monitor 都独立
"""
import numpy as np
import pandas as pd
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass, field
from enum import Enum
import warnings
warnings.filterwarnings('ignore')
class MarketRegime(Enum):
STRONG_TREND_UP = "强趋势上涨"
STRONG_TREND_DOWN = "强趋势下跌"
HIGH_VOL_SHAKE = "高波动剧烈震荡"
LOW_VOL_STABLE = "低波动平稳震荡"
CHAOTIC = "混乱无序"
UNKNOWN = "无法判断"
class StrategyType(Enum):
TREND_FOLLOWING = "趋势跟踪做T"
GRID_TRADING = "网格交易做T"
BOLLINGER_REVERSION = "布林带回归做T"
NO_TRADE = "暂停交易"
@dataclass
class MarketDiagnosis:
regime: MarketRegime
r_squared: float
volatility: float
adf_pvalue: float
recommended_strategy: StrategyType
strategy_params: Dict
confidence: float
reasoning: str
class IntradayStrategySelector:
"""
根据 5min K 线自动判别市场状态 + 推荐日内做T 策略
用法:
selector = IntradayStrategySelector()
diagnosis = selector.diagnose(df_5min, open_gap_pct=0.3)
print(diagnosis.recommended_strategy)
"""
def __init__(self,
trend_r2_threshold: float = 0.75,
chaos_r2_threshold: float = 0.30,
adf_significance: float = 0.05,
vol_lookback: int = 20,
high_vol_threshold: float = 0.30,
grid_count: int = 3,
bb_period: int = 20,
bb_std: float = 2.0,
ema_period: int = 5):
self.trend_r2_threshold = trend_r2_threshold
self.chaos_r2_threshold = chaos_r2_threshold
self.adf_significance = adf_significance
self.vol_lookback = vol_lookback
self.high_vol_threshold = high_vol_threshold
self.grid_count = grid_count
self.bb_period = bb_period
self.bb_std = bb_std
self.ema_period = ema_period
def diagnose(self, df: pd.DataFrame, open_gap_pct: float = 0.0) -> MarketDiagnosis:
if len(df) < 10:
raise ValueError(f"需要至少 10 根 K 线, 拿到 {len(df)}")
for col in ['open', 'high', 'low', 'close']:
if col not in df.columns:
raise ValueError(f"df 缺少 '{col}' 列")
prices = df['close'].values
r_squared = self._calculate_r_squared(prices)
volatility = self._calculate_historical_volatility(prices)
adf_pvalue = self._adf_test(prices)
slope = self._calculate_trend_slope(prices)
regime, confidence, reasoning = self._classify_regime(
r_squared, volatility, adf_pvalue, slope, open_gap_pct
)
strategy, params = self._match_strategy(regime, df, volatility, r_squared)
return MarketDiagnosis(
regime=regime,
r_squared=round(r_squared, 4),
volatility=round(volatility, 4),
adf_pvalue=round(adf_pvalue, 4),
recommended_strategy=strategy,
strategy_params=params,
confidence=round(confidence, 2),
reasoning=reasoning,
)
def _calculate_r_squared(self, prices: np.ndarray) -> float:
n = len(prices)
if n < 2:
return 0.0
x = np.arange(1, n + 1)
y = prices
x_mean = np.mean(x)
y_mean = np.mean(y)
numerator = np.sum((x - x_mean) * (y - y_mean))
denominator = np.sqrt(np.sum((x - x_mean) ** 2) * np.sum((y - y_mean) ** 2))
if denominator == 0:
return 0.0
r = numerator / denominator
return r ** 2
def _calculate_historical_volatility(self, prices: np.ndarray) -> float:
if len(prices) < 2:
return 0.0
log_returns = np.diff(np.log(prices))
return float(np.std(log_returns) * np.sqrt(252))
def _calculate_trend_slope(self, prices: np.ndarray) -> float:
n = len(prices)
if n < 2:
return 0.0
x = np.arange(1, n + 1)
y = prices
x_mean = np.mean(x)
y_mean = np.mean(y)
slope = np.sum((x - x_mean) * (y - y_mean)) / np.sum((x - x_mean) ** 2)
return float(slope / y_mean) if y_mean != 0 else 0.0
def _adf_test(self, prices: np.ndarray) -> float:
"""简化 ADF (用一阶差分自相关近似)
生产建议用 statsmodels.tsa.stattools.adfuller
"""
try:
from statsmodels.tsa.stattools import adfuller
result = adfuller(prices, autolag='AIC')
return float(result[1])
except ImportError:
# Fallback: 用一阶差分自相关近似
diffs = np.diff(prices)
if len(diffs) < 10:
return 1.0
autocorr = float(np.corrcoef(diffs[:-1], diffs[1:])[0, 1])
if abs(autocorr) < 0.1:
return 0.01
elif abs(autocorr) < 0.3:
return 0.05
elif abs(autocorr) < 0.5:
return 0.15
else:
return 0.50
def _classify_regime(self, r2: float, vol: float, adf_p: float,
slope: float, gap: float) -> Tuple[MarketRegime, float, str]:
if r2 > self.trend_r2_threshold:
regime = MarketRegime.STRONG_TREND_UP if slope > 0.002 else MarketRegime.STRONG_TREND_DOWN
confidence = min(r2, 1.0)
reasoning = (f"R²={r2:.3f}>0.75, 市场呈现强趋势状态。"
f"线性回归斜率={slope:.4f}, 方向明确。"
f"此类行情适合顺势做T,严禁逆势网格。")
return regime, confidence, reasoning
if r2 < self.chaos_r2_threshold:
if adf_p < self.adf_significance:
if vol > self.high_vol_threshold:
regime = MarketRegime.HIGH_VOL_SHAKE
confidence = 0.70
reasoning = (f"R²={r2:.3f}<0.30, ADF p={adf_p:.3f}<0.05, "
f"但波动率={vol:.2%}偏高。市场为高波动震荡,"
f"适宜宽间距的逆势策略,需严格止损。")
else:
regime = MarketRegime.LOW_VOL_STABLE
confidence = 0.85
reasoning = (f"R²={r2:.3f}<0.30, ADF p={adf_p:.3f}<0.05, "
f"波动率={vol:.2%}适中。经典震荡市,"
f"是网格和布林带回归策略的理想环境。")
else:
regime = MarketRegime.CHAOTIC
confidence = 0.40
reasoning = (f"R²={r2:.3f}<0.30, 但 ADF p={adf_p:.3f}>0.05, "
f"价格不具均值回归特性,属混乱状态,建议观望。")
return regime, confidence, reasoning
regime = MarketRegime.UNKNOWN
confidence = 0.30
reasoning = (f"R²={r2:.3f} 处于过渡区间(0.30-0.75), "
f"市场方向不明。建议等待模式清晰后再交易。")
return regime, confidence, reasoning
def _match_strategy(self, regime: MarketRegime, df: pd.DataFrame,
vol: float, r2: float) -> Tuple[StrategyType, Dict]:
current_price = float(df['close'].iloc[-1])
params = {}
if regime == MarketRegime.STRONG_TREND_UP:
strategy = StrategyType.TREND_FOLLOWING
ema = float(df['close'].ewm(span=self.ema_period).mean().iloc[-1])
params = {
"direction": "long_only",
"entry_trigger": f"价格回踩 {ema:.2f} (EMA{self.ema_period}) 不破",
"stop_loss": f"{ema * 0.995:.2f}",
"take_profit": f"{current_price * 1.02:.2f}",
}
elif regime == MarketRegime.STRONG_TREND_DOWN:
strategy = StrategyType.TREND_FOLLOWING
ema = float(df['close'].ewm(span=self.ema_period).mean().iloc[-1])
params = {
"direction": "short_only",
"entry_trigger": f"价格反弹至 {ema:.2f} (EMA{self.ema_period}) 受阻",
"stop_loss": f"{ema * 1.005:.2f}",
"take_profit": f"{current_price * 0.98:.2f}",
}
elif regime == MarketRegime.LOW_VOL_STABLE:
strategy = StrategyType.GRID_TRADING
avg_amplitude = float(((df['high'] - df['low']) / df['close']).mean())
grid_spacing = max(avg_amplitude * 0.8, 0.005)
params = {
"grid_spacing": f"{grid_spacing:.3%}",
"grid_levels": self.grid_count,
"base_price": f"{current_price:.2f}",
"reverse_at_boundary": True,
}
elif regime == MarketRegime.HIGH_VOL_SHAKE:
strategy = StrategyType.BOLLINGER_REVERSION
rolling_std = float(df['close'].rolling(self.bb_period).std().iloc[-1])
ma = float(df['close'].rolling(self.bb_period).mean().iloc[-1])
upper = ma + self.bb_std * rolling_std
lower = ma - self.bb_std * rolling_std
params = {
"upper_band": f"{upper:.2f}",
"lower_band": f"{lower:.2f}",
"sell_at_upper": True,
"buy_at_lower": True,
"stop_if_break": True,
}
else:
strategy = StrategyType.NO_TRADE
params = {"reason": "市场状态不清晰,等待趋势或明确震荡信号"}
return strategy, params
def diagnose_market(df: pd.DataFrame, open_gap_pct: float = 0.0) -> MarketDiagnosis:
"""便捷函数"""
return IntradayStrategySelector().diagnose(df, open_gap_pct)
if __name__ == '__main__':
np.random.seed(42)
# 场景 1: 震荡市
n = 25
base = 10.0
noise = np.random.randn(n) * 0.05
close = base + noise
high = close + np.abs(np.random.randn(n) * 0.03)
low = close - np.abs(np.random.randn(n) * 0.03)
df = pd.DataFrame({
'open': close - 0.01,
'high': high,
'low': low,
'close': close,
'volume': np.random.randint(1000, 5000, n),
})
diag = diagnose_market(df, open_gap_pct=0.0)
print(f"场景 1 (震荡市): {diag.regime.value} | R²={diag.r_squared} | 策略: {diag.recommended_strategy.value}")
print(f" 推理: {diag.reasoning}\n")
# 场景 2: 强趋势
close2 = base + np.cumsum(np.random.randn(n) * 0.02) * 2 # 上升趋势
high2 = close2 + 0.05
low2 = close2 - 0.05
df2 = pd.DataFrame({
'open': close2 - 0.01,
'high': high2,
'low': low2,
'close': close2,
'volume': np.random.randint(1000, 5000, n),
})
diag2 = diagnose_market(df2, open_gap_pct=0.5)
print(f"场景 2 (趋势市): {diag2.regime.value} | R²={diag2.r_squared} | 策略: {diag2.recommended_strategy.value}")
print(f" 推理: {diag2.reasoning}")
print(f" 参数: {diag2.strategy_params}")
intraday-regime-detector/scripts/regime_scan.py
+186
View File
@@ -0,0 +1,186 @@
#!/usr/bin/env python3
"""
regime_scan.py - 扫描港美股日内候选的市场状态 + 推荐策略
不交易, 只判别 + 推 QQ
"""
import sys
import json
import subprocess
import re
from pathlib import Path
sys.path.insert(0, '/home/openclaw/.hermes/skills/trading/intraday-regime-detector/scripts')
from intraday_regime import IntradayStrategySelector, MarketRegime, StrategyType
CANDIDATE_HK = Path('/home/openclaw/.hermes/skills/trading/quant-factor-mining/artifacts/hk_intraday_latest.json')
CANDIDATE_US = Path('/home/openclaw/.hermes/skills/trading/quant-factor-mining/artifacts/us_intraday_latest.json')
def fetch_klines_hk(symbol: str, period: str = '5m', count: int = 30) -> list:
"""港股表格 parser"""
result = subprocess.run(
['proxychains4', '-f', '/home/openclaw/.proxychains/proxychains.conf',
'/home/openclaw/.local/bin/longbridge', '--profile', 'lb_real',
'candlesticks', symbol, period, '--count', str(count)],
capture_output=True, text=True, timeout=30,
)
klines = []
pattern = re.compile(
r'│\s*(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2})\s*│'
r'\s*([\d,.]+)\s*│\s*([\d,.]+)\s*│\s*([\d,.]+)\s*│\s*([\d,.]+)\s*│\s*([\d,.]+)\s*│'
)
for line in result.stdout.split('\n'):
m = pattern.search(line)
if m:
ts, o, h, l, c, v = m.groups()
def parse_num(s):
return float(s.replace(',', ''))
klines.append({
'open': parse_num(o),
'high': parse_num(h),
'low': parse_num(l),
'close': parse_num(c),
'volume': parse_num(v),
})
return klines
def fetch_klines_us(symbol: str, period: str = '5m', count: int = 30) -> list:
"""美股 JSON"""
result = subprocess.run(
['proxychains4', '-f', '/home/openclaw/.proxychains/proxychains.conf',
'/home/openclaw/.local/bin/longbridge', '--profile', 'lb_real',
'candlesticks', symbol, period, '--count', str(count), '--json'],
capture_output=True, text=True, timeout=30,
)
start = result.stdout.find('[')
if start == -1:
return []
try:
data = json.loads(result.stdout[start:])
return [{
'open': float(k['open']),
'high': float(k['high']),
'low': float(k['low']),
'close': float(k['close']),
'volume': float(k.get('volume', 0)),
} for k in data if 'close' in k]
except Exception:
return []
def fetch_quote(symbol: str) -> dict:
result = subprocess.run(
['proxychains4', '-f', '/home/openclaw/.proxychains/proxychains.conf',
'/home/openclaw/.local/bin/longbridge', '--profile', 'lb_real',
'quote', symbol, '--json'],
capture_output=True, text=True, timeout=30,
)
text = result.stdout
start = text.find('[')
if start == -1:
return {}
try:
return json.loads(text[start:])[0]
except Exception:
return {}
def analyze_market(symbol: str, market: str, klines: list, quote: dict) -> str:
"""返回单支票分析报告"""
if not klines or not quote:
return f"❌ {symbol} 数据缺失"
try:
import pandas as pd
df = pd.DataFrame(klines)
except ImportError:
return f"❌ pandas 未装"
prev_close = quote.get('prev_close', 0)
current_price = quote['last_done']
open_p = quote['open']
gap_pct = ((open_p - prev_close) / prev_close * 100) if prev_close else 0
selector = IntradayStrategySelector()
diag = selector.diagnose(df, open_gap_pct=gap_pct)
# 策略 emoji
strategy_emoji = {
StrategyType.TREND_FOLLOWING: '📈',
StrategyType.GRID_TRADING: '🔲',
StrategyType.BOLLINGER_REVERSION: '📊',
StrategyType.NO_TRADE: '⛔',
}
regime_short = {
MarketRegime.STRONG_TREND_UP: '强趋↑',
MarketRegime.STRONG_TREND_DOWN: '强趋↓',
MarketRegime.HIGH_VOL_SHAKE: '高波震荡',
MarketRegime.LOW_VOL_STABLE: '低波震荡',
MarketRegime.CHAOTIC: '混乱',
MarketRegime.UNKNOWN: '未知',
}
params_str = '\n'.join(f" {k}: {v}" for k, v in diag.strategy_params.items())
return (
f"\n{strategy_emoji.get(diag.recommended_strategy, '•')} **{symbol}** ({market}) "
f"现价 ${current_price:.2f} ({gap_pct:+.2f}%) "
f"置信度 {diag.confidence:.0%}\n"
f" 状态: {regime_short.get(diag.regime, diag.regime.value)} | "
f"R²={diag.r_squared} | 波动率={diag.volatility:.2%} | ADF p={diag.adf_pvalue}\n"
f" 推荐: {diag.recommended_strategy.value}\n"
f"{params_str}"
)
def scan_market(market: str, candidate_file: Path, fetch_klines_func) -> list:
"""扫描一个市场"""
if not candidate_file.exists():
return [f"⚠️ 候选池不存在: {candidate_file.name}"]
with open(candidate_file) as f:
data = json.load(f)
results = data.get('results', [])[:5] # top 5
date = data.get('date', '?')[:10]
if not results:
return [f"⚠️ {market} 候选池为空"]
reports = [f"📊 {market} 日内市场状态扫描 ({date})"]
for entry in results:
symbol = entry['ticker']
score = entry['score']
try:
quote = fetch_quote(symbol)
klines = fetch_klines_func(symbol, '5m', 30)
report = analyze_market(symbol, market, klines, quote)
reports.append(report)
except Exception as e:
reports.append(f"❌ {symbol} 异常: {e}")
return reports
def main():
# 港股 + 美股
hk_reports = scan_market('HK', CANDIDATE_HK, fetch_klines_hk)
us_reports = scan_market('US', CANDIDATE_US, fetch_klines_us)
print(f"📊 日内市场状态扫描 ({hk_reports[0].split('(')[-1].rstrip(')')})\n")
print('=' * 60)
print('\n--- 港股 ---')
for r in hk_reports[1:]:
print(r)
print()
print('\n--- 美股 ---')
for r in us_reports[1:]:
print(r)
if __name__ == '__main__':
main()
strategy-management/scripts/exit_levels.py
+292
View File
@@ -0,0 +1,292 @@
"""
exit_levels.py - 港美股日内做T 出场点位计算 (混合公式)
设计: 方法 2 (百分比波动率) + 方法 3 (关键价位) 混合
- 关键位: day_high/low / prev_high/low / VWAP (从 indicators.py)
- 波动率: ATR (从 indicators.py)
- R:R 强制下限 1.5, 不达标信号否决
⚠️ 这是港美股做T专用, 币圈用 crypto-t-monitor 的 ATR 公式 (独立)
用法:
from exit_levels import calc_exit_levels
result = calc_exit_levels(
entry=100.0,
atr=5.0,
current_price=100.0,
day_high=103.0,
day_low=97.0,
prev_high=106.0,
prev_low=94.0,
vwap=101.0,
side='long',
min_rr=1.5,
)
if result is None:
print("信号否决: R:R 不达标")
else:
sl, tp1, tp2 = result
print(f"SL={sl} TP1={tp1} TP2={tp2}")
"""
from dataclasses import dataclass
from typing import Optional, Literal
@dataclass
class ExitLevels:
"""出场点位结果"""
sl: float # 止损价位
tp1: float # 第一止盈 (半平)
tp2: float # 第二止盈 (全平)
entry: float # 入场价 (回填, 方便调用方记录)
side: str # 'long' / 'short'
risk: float # 风险 (entry - SL)
reward: float # 奖励 (TP1 - entry)
rr_ratio: float # R:R (reward/risk)
sl_method: str # 'vol_pct' | 'key_level' | 'hybrid'
tp_method: str # 'vol_pct' | 'key_level' | 'hybrid'
note: str = "" # 备注 (VWAP 锁定等)
def calc_exit_levels(
entry: float,
atr: float,
current_price: float,
day_high: Optional[float] = None,
day_low: Optional[float] = None,
prev_high: Optional[float] = None,
prev_low: Optional[float] = None,
vwap: Optional[float] = None,
side: Literal['long', 'short'] = 'long',
min_rr: float = 1.5,
# 方法 2 权重 (百分比波动率)
vol_sl_multi: float = 1.0, # SL 距离 = vol × sl_multi
vol_tp1_multi: float = 2.0, # TP1 距离 = vol × tp1_multi (默认 2.0 倍 SL → R:R 2:1)
vol_tp2_multi: float = 3.0, # TP2 距离 = vol × tp2_multi
# 方法 3 权重 (关键位) - 离关键位的 buffer
key_level_buffer_pct: float = 0.001, # 0.1% 缓冲 (避免瞬时触发)
) -> Optional[ExitLevels]:
"""
计算 SL/TP1/TP2 (混合公式: 波动率 + 关键位)
Args:
entry: 入场价 (假设已知, 或用 adjust_to_ask1 拿到的价)
atr: 当前 K 线 ATR (14 周期, 从 indicators.atr())
current_price: 当前实时价 (用于 vol_pct 计算)
day_high/low: 今日最高/最低 (从 longbridge quote)
prev_high/low: 昨日最高/最低 (从 longbridge K 线)
vwap: 成交量加权平均价 (从 indicators.vwap())
side: 'long' (做多) / 'short' (做空)
min_rr: 最小 R:R (默认 1.5)
vol_sl_multi / tp1_multi / tp2_multi: ATR 倍数
key_level_buffer_pct: 关键位 buffer (避免价格精确等于关键位)
Returns:
ExitLevels 或 None (R:R 不达标)
Examples:
>>> calc_exit_levels(entry=100, atr=5, current_price=100,
... day_high=115, day_low=92, prev_high=118, prev_low=90,
... vwap=105, side='long')
ExitLevels(sl=91.2, tp1=114.2, tp2=120.0, ...)
"""
if entry <= 0 or atr <= 0 or current_price <= 0:
raise ValueError("entry/atr/current_price must be > 0")
# === 方法 2: 百分比波动率 (基于 ATR) ===
vol_pct = atr / current_price
vol_sl_dist = vol_pct * vol_sl_multi
vol_tp1_dist = vol_pct * vol_tp1_multi
vol_tp2_dist = vol_pct * vol_tp2_multi
if side == 'long':
sl_vol = entry * (1 - vol_sl_dist)
tp1_vol = entry * (1 + vol_tp1_dist)
tp2_vol = entry * (1 + vol_tp2_dist)
else:
sl_vol = entry * (1 + vol_sl_dist)
tp1_vol = entry * (1 - vol_tp1_dist)
tp2_vol = entry * (1 - vol_tp2_dist)
# === 方法 3: 关键价位 ===
# 多仓: SL 取 entry **下方**的支撑; TP1 取 entry **上方**的阻力
# 空仓: SL 取 entry **上方**的阻力; TP1 取 entry **下方**的支撑
sl_key = None
tp1_key = None
note = ""
if side == 'long':
# SL 关键位 (entry 下方)
sl_candidates = []
if day_low is not None and day_low < entry:
sl_candidates.append(day_low * (1 - key_level_buffer_pct))
if prev_low is not None and prev_low < entry:
sl_candidates.append(prev_low * (1 - key_level_buffer_pct))
if vwap is not None and vwap < entry:
sl_candidates.append(vwap * (1 - key_level_buffer_pct))
sl_key = max(sl_candidates) if sl_candidates else None
# TP1 关键位 (entry 上方)
tp1_candidates = []
if day_high is not None and day_high > entry:
tp1_candidates.append(day_high * (1 - key_level_buffer_pct))
if prev_high is not None and prev_high > entry:
tp1_candidates.append(prev_high * (1 - key_level_buffer_pct))
if vwap is not None and vwap > entry:
tp1_candidates.append(vwap * (1 - key_level_buffer_pct))
tp1_key = min(tp1_candidates) if tp1_candidates else None
else: # short
# SL 关键位 (entry 上方, 空仓止损 = 价格涨到这里平)
sl_candidates = []
if day_high is not None and day_high > entry:
sl_candidates.append(day_high * (1 + key_level_buffer_pct))
if prev_high is not None and prev_high > entry:
sl_candidates.append(prev_high * (1 + key_level_buffer_pct))
if vwap is not None and vwap > entry:
sl_candidates.append(vwap * (1 + key_level_buffer_pct))
sl_key = min(sl_candidates) if sl_candidates else None
# TP1 关键位 (entry 下方)
tp1_candidates = []
if day_low is not None and day_low < entry:
tp1_candidates.append(day_low * (1 + key_level_buffer_pct))
if prev_low is not None and prev_low < entry:
tp1_candidates.append(prev_low * (1 + key_level_buffer_pct))
if vwap is not None and vwap < entry:
tp1_candidates.append(vwap * (1 + key_level_buffer_pct))
tp1_key = max(tp1_candidates) if tp1_candidates else None
# === 混合: vol_pct 主导 (70%), 关键位微调 (30%) ===
# 关键位 30% 权重, 防止 VWAP 等动态位锁死
# vol_pct 至少占 70% (最终值不会偏离 vol_pct 太远)
if side == 'long':
if sl_key is not None:
SL = sl_vol * 0.7 + sl_key * 0.3
sl_method = 'hybrid_blend'
else:
SL = sl_vol
sl_method = 'vol_pct'
if tp1_key is not None:
TP1 = tp1_vol * 0.7 + tp1_key * 0.3
tp_method = 'hybrid_blend'
else:
TP1 = tp1_vol
tp_method = 'vol_pct'
TP2 = tp2_vol
else: # short
if sl_key is not None:
SL = sl_vol * 0.7 + sl_key * 0.3
sl_method = 'hybrid_blend'
else:
SL = sl_vol
sl_method = 'vol_pct'
if tp1_key is not None:
TP1 = tp1_vol * 0.7 + tp1_key * 0.3
tp_method = 'hybrid_blend'
else:
TP1 = tp1_vol
tp_method = 'vol_pct'
TP2 = tp2_vol
# === R:R 检查 ===
if side == 'long':
risk = entry - SL
reward = TP1 - entry
else:
risk = SL - entry
reward = entry - TP1
if risk <= 0:
return None # 止损 >= 入场 (逻辑错误)
rr_ratio = reward / risk if risk > 0 else 0
# 风险 vs reward
if abs(reward) < min_rr * abs(risk):
# R:R 不达标
if sl_key == tp1_key and sl_key is not None:
note = f"VWAP 既作支撑又作阻力, 价格窄幅震荡 (SL=TP1={sl_key:.2f})"
else:
note = f"R:R {rr_ratio:.2f} < {min_rr}, 信号否决"
return None
return ExitLevels(
sl=SL,
tp1=TP1,
tp2=TP2,
entry=entry,
side=side,
risk=abs(risk),
reward=abs(reward),
rr_ratio=rr_ratio,
sl_method=sl_method,
tp_method=tp_method,
note=note,
)
def format_levels(levels: ExitLevels) -> str:
"""格式化输出 (QQ 推送用)"""
side_emoji = '🟢' if levels.side == 'long' else '🔴'
return (
f"{side_emoji} {levels.side.upper()} @ ${levels.entry:.2f}\n"
f" SL: ${levels.sl:.2f} ({levels.sl_method})\n"
f" TP1: ${levels.tp1:.2f} ({levels.tp_method})\n"
f" TP2: ${levels.tp2:.2f}\n"
f" Risk/Reward: 1:{levels.rr_ratio:.2f}"
)
if __name__ == '__main__':
# 自检: 用 indicators.py 的真实数据测试
print("=" * 60)
print("exit_levels.py - 自检")
print("=" * 60)
# 案例 1: NVDA 高波动 (有 R:R)
print("\n[案例 1] NVDA $100, ATR=$8, day H/L=$115/$92, prev H/L=$118/$90")
result = calc_exit_levels(
entry=100.0, atr=8.0, current_price=100.0,
day_high=115.0, day_low=92.0,
prev_high=118.0, prev_low=90.0,
vwap=105.0,
side='long',
)
if result:
print(format_levels(result))
else:
print("❌ 信号否决")
# 案例 2: 价在 VWAP 上下窄幅震荡
print("\n[案例 2] NVDA $100, ATR=$3, VWAP=$101 (紧贴)")
result = calc_exit_levels(
entry=100.0, atr=3.0, current_price=100.0,
day_high=103.0, day_low=98.0,
prev_high=105.0, prev_low=95.0,
vwap=101.0,
side='long',
)
if result:
print(format_levels(result))
else:
print("❌ 信号否决 (R:R 不达标 / VWAP 锁定)")
# 案例 3: 空仓 + 大波动
print("\n[案例 3] TSDA short $200, ATR=$12")
result = calc_exit_levels(
entry=200.0, atr=12.0, current_price=200.0,
day_high=212.0, day_low=188.0,
prev_high=215.0, prev_low=185.0,
vwap=205.0,
side='short',
)
if result:
print(format_levels(result))
else:
print("❌ 信号否决")