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

feat: 初始化量化交易知识库 v1.0

浏览代码 
- 01_基础理论：量化交易基础概念、市场微观结构、加密货币特殊性
- 02_技术指标：完整指标体系（MA/EMA/MACD/RSI/KDJ/布林带/SuperTrend/DMI等）
- 03_交易策略：趋势跟踪、均值回归、套利、动量策略详解
- 04_交易信号系统：多指标共振评分引擎（基于 tradehk 项目）
- 05_市场品种：加密货币、XAUT黄金代币、代币化美股全览
- 06_数据流程：数据采集、清洗、存储、实时流处理
- 07_回测框架：回测方法论、偏差规避、绩效评估指标
- 08_风险管理：仓位管理、止损止盈、Kelly公式、杠杆管理
- 09_AI与机器学习：深度学习、强化学习、LLM在量化投资中的应用
- 10_链上数据分析：SOPR/MVRV/巨鲸监控/衍生品数据
- 11_参考文献：arXiv论文汇总、开源项目、数据平台资源
- samples/：Python信号计算器和回测样本代码

参考项目：tradehk（ssh://git@git.hk.hao.work:2222/hao/tradehk.git）
全部中文化，适用于加密货币（CEX/DEX）、XAUT黄金、代币化美股
这个提交包含在：
Manus Quant Agent
2026-03-05 21:36:56 -05:00
当前提交 f1d939b460
共修改 15 个文件，包含 3431 行新增和 0 行删除
下载 Patch 文件 下载 Diff 文件 展开所有文件 折叠所有文件

280
samples/backtest_sample.py 普通文件
查看文件

@@ -0,0 +1,280 @@
"""
量化交易回测样本 - MACD + EWO 趋势跟踪策略
基于 tradehk 项目的信号系统进行历史回测
使用方法：
pip install pandas numpy matplotlib requests
python backtest_sample.py
"""
import pandas as pd
import numpy as np
import requests
import json
from datetime import datetime
# ============================================================
# 数据获取
# ============================================================
def fetch_binance_klines(symbol: str, interval: str, limit: int = 1000) -> pd.DataFrame:
url = "https://api.binance.com/api/v3/klines"
params = {"symbol": symbol, "interval": interval, "limit": limit}
resp = requests.get(url, params=params, timeout=10)
resp.raise_for_status()
data = resp.json()
df = pd.DataFrame(data, 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 calc_ema(s, n): return s.ewm(span=n, adjust=False).mean()
def calc_sma(s, n): return s.rolling(n).mean()
def calc_rma(s, n): return s.ewm(alpha=1/n, adjust=False).mean()
def calc_macd(close, fast=10, slow=20, signal=10):
macd = calc_ema(close, fast) - calc_ema(close, slow)
sig = calc_ema(macd, signal)
return macd, sig, macd - sig
def calc_ewo(close):
return calc_ema(close, 5) - calc_ema(close, 35)
def calc_ao(df):
mid = (df['high'] + df['low']) / 2
return calc_sma(mid, 5) - calc_sma(mid, 34)
def calc_rsi(close, period=14):
delta = close.diff()
gain = calc_rma(delta.clip(lower=0), period)
loss = calc_rma((-delta).clip(lower=0), period)
rs = gain / loss.replace(0, np.nan)
return 100 - 100 / (1 + rs)
def calc_atr(df, period=14):
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 calc_rma(tr, period)
# ============================================================
# 策略：MACD 金叉/死叉 + EWO 大方向过滤
# ============================================================
def generate_signals(df: pd.DataFrame) -> pd.DataFrame:
"""
策略逻辑：
- 大方向：EWO > 0 看多，EWO < 0 看空
- 入场：MACD 金叉（大方向看多时）或死叉（大方向看空时）
- 止损：ATR 动态止损（2 × ATR）
"""
close = df['close']
macd_line, signal_line, histogram = calc_macd(close)
ewo = calc_ewo(close)
ao = calc_ao(df)
rsi = calc_rsi(close)
atr = calc_atr(df)
ma10 = calc_sma(close, 10)
ma100 = calc_sma(close, 100)
signals = pd.DataFrame(index=df.index)
signals['close'] = close
signals['macd'] = macd_line
signals['signal'] = signal_line
signals['histogram'] = histogram
signals['ewo'] = ewo
signals['ao'] = ao
signals['rsi'] = rsi
signals['atr'] = atr
signals['ma10'] = ma10
signals['ma100'] = ma100
# MACD 金叉/死叉
signals['macd_cross_up'] = (macd_line > signal_line) & (macd_line.shift(1) <= signal_line.shift(1))
signals['macd_cross_down'] = (macd_line < signal_line) & (macd_line.shift(1) >= signal_line.shift(1))
# 大方向过滤（EWO）
signals['trend_bullish'] = ewo > 0
signals['trend_bearish'] = ewo < 0
# 最终信号
signals['buy_signal'] = signals['macd_cross_up'] & signals['trend_bullish']
signals['sell_signal'] = signals['macd_cross_down'] & signals['trend_bearish']
return signals.dropna()
# ============================================================
# 回测引擎
# ============================================================
def backtest(df: pd.DataFrame, initial_capital: float = 10000, commission: float = 0.001) -> dict:
"""
简单回测引擎
- 固定仓位：每次使用全部资金
- 止损：2 × ATR
- 止盈：4 × ATR（2:1 盈亏比）
"""
signals = generate_signals(df)
capital = initial_capital
position = 0 # 0=空仓, 1=多头
entry_price = 0
stop_loss = 0
take_profit = 0
trades = []
equity_curve = [capital]
for i in range(1, len(signals)):
row = signals.iloc[i]
prev_row = signals.iloc[i-1]
# 检查止损/止盈
if position == 1:
if row['close'] <= stop_loss:
# 止损出场
pnl = (row['close'] - entry_price) * (capital / entry_price)
pnl -= abs(pnl) * commission * 2
capital += pnl
trades.append({
'type': '止损出场',
'entry': entry_price,
'exit': row['close'],
'pnl': pnl,
'pnl_pct': (row['close'] - entry_price) / entry_price * 100
})
position = 0
elif row['close'] >= take_profit:
# 止盈出场
pnl = (row['close'] - entry_price) * (capital / entry_price)
pnl -= abs(pnl) * commission * 2
capital += pnl
trades.append({
'type': '止盈出场',
'entry': entry_price,
'exit': row['close'],
'pnl': pnl,
'pnl_pct': (row['close'] - entry_price) / entry_price * 100
})
position = 0
elif row['sell_signal']:
# 信号反转出场
pnl = (row['close'] - entry_price) * (capital / entry_price)
pnl -= abs(pnl) * commission * 2
capital += pnl
trades.append({
'type': '信号出场',
'entry': entry_price,
'exit': row['close'],
'pnl': pnl,
'pnl_pct': (row['close'] - entry_price) / entry_price * 100
})
position = 0
# 开仓
if position == 0 and row['buy_signal']:
entry_price = row['close']
atr = row['atr']
stop_loss = entry_price - 2 * atr
take_profit = entry_price + 4 * atr
position = 1
capital -= capital * commission # 入场手续费
# 记录净值
if position == 1:
unrealized = (row['close'] - entry_price) * (capital / entry_price)
equity_curve.append(capital + unrealized)
else:
equity_curve.append(capital)
# 计算绩效指标
equity = pd.Series(equity_curve)
returns = equity.pct_change().dropna()
total_return = (equity.iloc[-1] / initial_capital - 1) * 100
max_drawdown = ((equity.cummax() - equity) / equity.cummax()).max() * 100
if returns.std() > 0:
sharpe = returns.mean() / returns.std() * np.sqrt(365 * 24)
else:
sharpe = 0
winning_trades = [t for t in trades if t['pnl'] > 0]
losing_trades = [t for t in trades if t['pnl'] <= 0]
win_rate = len(winning_trades) / len(trades) if trades else 0
avg_win = np.mean([t['pnl_pct'] for t in winning_trades]) if winning_trades else 0
avg_loss = np.mean([t['pnl_pct'] for t in losing_trades]) if losing_trades else 0
profit_factor = abs(avg_win / avg_loss) if avg_loss != 0 else 0
return {
'初始资金': f"${initial_capital:,.2f}",
'最终资金': f"${equity.iloc[-1]:,.2f}",
'总收益率': f"{total_return:.2f}%",
'最大回撤': f"{max_drawdown:.2f}%",
'夏普比率': f"{sharpe:.2f}",
'总交易次数': len(trades),
'胜率': f"{win_rate:.2%}",
'平均盈利': f"{avg_win:.2f}%",
'平均亏损': f"{avg_loss:.2f}%",
'盈亏比': f"{profit_factor:.2f}",
'止损次数': len([t for t in trades if t['type'] == '止损出场']),
'止盈次数': len([t for t in trades if t['type'] == '止盈出场']),
'信号出场次数': len([t for t in trades if t['type'] == '信号出场']),
}
# ============================================================
# 主程序
# ============================================================
if __name__ == '__main__':
print("=" * 60)
print("量化策略回测 - MACD + EWO 趋势跟踪")
print("策略：MACD 金叉（EWO 多头时）做多，ATR 动态止损止盈")
print("=" * 60)
test_cases = [
('BTCUSDT', '4h', 'BTC/USDT 4小时'),
('ETHUSDT', '4h', 'ETH/USDT 4小时'),
]
for symbol, interval, label in test_cases:
print(f"\n{'─' * 50}")
print(f"回测品种：{label}")
print(f"{'─' * 50}")
try:
df = fetch_binance_klines(symbol, interval, limit=1000)
print(f"数据范围：{df.index[0].strftime('%Y-%m-%d')} 至 {df.index[-1].strftime('%Y-%m-%d')}")
print(f"K 线数量：{len(df)}")
results = backtest(df, initial_capital=10000)
print("\n📊 回测结果：")
for key, value in results.items():
print(f" {key:12s}: {value}")
except Exception as e:
print(f"错误：{e}")
print(f"\n{'=' * 60}")
print("⚠️ 免责声明：以上回测结果仅供学习参考，不构成投资建议")
print(" 历史表现不代表未来收益，实盘交易存在亏损风险")
print("=" * 60)

374
samples/signal_calculator_sample.py 普通文件
查看文件

@@ -0,0 +1,374 @@
"""
量化交易信号计算器 - 完整示例
基于 tradehk 项目的多指标共振信号系统 Python 实现
使用方法：
pip install pandas numpy requests
python signal_calculator_sample.py
"""
import pandas as pd
import numpy as np
import requests
from typing import Optional, Tuple
from dataclasses import dataclass
# ============================================================
# 数据获取
# ============================================================
def fetch_binance_klines(symbol: str, interval: str, limit: int = 500) -> pd.DataFrame:
"""
从 Binance 获取 K 线数据
参数：
symbol: 交易对，如 'BTCUSDT'
interval: 时间周期，如 '1h', '4h', '1d'
limit: 获取数量（最大 1000）
"""
url = "https://api.binance.com/api/v3/klines"
params = {"symbol": symbol, "interval": interval, "limit": limit}
resp = requests.get(url, params=params, timeout=10)
resp.raise_for_status()
data = resp.json()
df = pd.DataFrame(data, 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 calc_rma(series: pd.Series, period: int) -> pd.Series:
"""Wilder 平滑移动均线（RMA）"""
alpha = 1.0 / period
return series.ewm(alpha=alpha, adjust=False).mean()
def calc_ema(series: pd.Series, period: int) -> pd.Series:
"""指数移动均线（EMA）"""
return series.ewm(span=period, adjust=False).mean()
def calc_sma(series: pd.Series, period: int) -> pd.Series:
"""简单移动均线（SMA）"""
return series.rolling(period).mean()
def calc_atr(df: pd.DataFrame, period: int = 14) -> pd.Series:
"""真实波动幅度（ATR）"""
high, low, close = df['high'], df['low'], df['close']
tr = pd.concat([
high - low,
(high - close.shift(1)).abs(),
(low - close.shift(1)).abs()
], axis=1).max(axis=1)
return calc_rma(tr, period)
def calc_rsi(close: pd.Series, period: int = 14) -> pd.Series:
"""相对强弱指数（RSI）"""
delta = close.diff()
gain = delta.clip(lower=0)
loss = (-delta).clip(lower=0)
avg_gain = calc_rma(gain, period)
avg_loss = calc_rma(loss, period)
rs = avg_gain / avg_loss.replace(0, np.nan)
return 100 - (100 / (1 + rs))
def calc_macd(close: pd.Series, fast: int = 10, slow: int = 20, signal: int = 10) -> Tuple[pd.Series, pd.Series, pd.Series]:
"""
MACD（tradehk 参数：10, 20, 10）
返回：(macd线, 信号线, 柱状图)
"""
ema_fast = calc_ema(close, fast)
ema_slow = calc_ema(close, slow)
macd_line = ema_fast - ema_slow
signal_line = calc_ema(macd_line, signal)
histogram = macd_line - signal_line
return macd_line, signal_line, histogram
def calc_ewo(close: pd.Series) -> pd.Series:
"""艾略特波浪振荡器（EWO = EMA5 - EMA35）"""
return calc_ema(close, 5) - calc_ema(close, 35)
def calc_ao(df: pd.DataFrame) -> pd.Series:
"""动量振荡器（AO）"""
midpoint = (df['high'] + df['low']) / 2
return calc_sma(midpoint, 5) - calc_sma(midpoint, 34)
def calc_kdj(df: pd.DataFrame, period: int = 9, k_smooth: int = 3, d_smooth: int = 3) -> Tuple[pd.Series, pd.Series, pd.Series]:
"""KDJ 指标"""
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=k_smooth - 1, adjust=False).mean()
d = k.ewm(com=d_smooth - 1, adjust=False).mean()
j = 3 * k - 2 * d
return k, d, j
def calc_bollinger_bands(close: pd.Series, period: int = 20, multiplier: float = 2.0) -> Tuple[pd.Series, pd.Series, pd.Series]:
"""布林带"""
middle = calc_sma(close, period)
std = close.rolling(period).std()
upper = middle + multiplier * std
lower = middle - multiplier * std
return upper, middle, lower
def calc_supertrend(df: pd.DataFrame, period: int = 10, multiplier: float = 3.0) -> Tuple[pd.Series, pd.Series]:
"""
超级趋势线（SuperTrend）
返回：(趋势值, 方向: 1=多头, -1=空头)
"""
atr = calc_atr(df, period)
hl2 = (df['high'] + df['low']) / 2
upper_band = hl2 + multiplier * atr
lower_band = hl2 - multiplier * atr
supertrend = pd.Series(np.nan, index=df.index)
direction = pd.Series(1, index=df.index)
for i in range(1, len(df)):
# 更新上轨
if upper_band.iloc[i] < upper_band.iloc[i-1] or df['close'].iloc[i-1] > upper_band.iloc[i-1]:
upper_band.iloc[i] = upper_band.iloc[i]
else:
upper_band.iloc[i] = upper_band.iloc[i-1]
# 更新下轨
if lower_band.iloc[i] > lower_band.iloc[i-1] or df['close'].iloc[i-1] < lower_band.iloc[i-1]:
lower_band.iloc[i] = lower_band.iloc[i]
else:
lower_band.iloc[i] = lower_band.iloc[i-1]
# 确定方向
if direction.iloc[i-1] == -1:
if df['close'].iloc[i] > upper_band.iloc[i]:
direction.iloc[i] = 1
else:
direction.iloc[i] = -1
else:
if df['close'].iloc[i] < lower_band.iloc[i]:
direction.iloc[i] = -1
else:
direction.iloc[i] = 1
supertrend.iloc[i] = lower_band.iloc[i] if direction.iloc[i] == 1 else upper_band.iloc[i]
return supertrend, direction
# ============================================================
# 信号生成（多指标共振）
# ============================================================
@dataclass
class SignalResult:
signal_type: str # 'BUY', 'SELL', 'NEUTRAL'
strength: str # 'STRONG', 'MODERATE', 'WEAK'
bullish_score: int
bearish_score: int
details: dict
def generate_signal(df: pd.DataFrame, use_kdj: bool = True, use_supertrend: bool = True) -> SignalResult:
"""
多指标共振信号生成器
参数：
df: 包含 OHLCV 数据的 DataFrame
use_kdj: 是否启用 KDJ 信号
use_supertrend: 是否启用 SuperTrend 信号
"""
close = df['close']
# 计算所有指标
ma10 = calc_sma(close, 10)
ma100 = calc_sma(close, 100)
macd_line, signal_line, histogram = calc_macd(close)
ao = calc_ao(df)
rsi = calc_rsi(close)
bb_upper, bb_middle, bb_lower = calc_bollinger_bands(close)
# 取最后两根 K 线
curr = df.index[-1]
prev = df.index[-2]
bullish_score = 0
bearish_score = 0
details = {}
# ---- 核心信号：MACD ----
macd_cross_up = macd_line[curr] > signal_line[curr] and macd_line[prev] <= signal_line[prev]
macd_cross_down = macd_line[curr] < signal_line[curr] and macd_line[prev] >= signal_line[prev]
if macd_cross_up:
bullish_score += 2
details['MACD'] = '金叉 (+2)'
elif macd_cross_down:
bearish_score += 2
details['MACD'] = '死叉 (+2)'
elif histogram[curr] > 0 and histogram[curr] > histogram[prev]:
bullish_score += 1
details['MACD'] = '柱状图扩大（正）(+1)'
elif histogram[curr] < 0 and histogram[curr] < histogram[prev]:
bearish_score += 1
details['MACD'] = '柱状图扩大（负）(+1)'
# ---- 核心信号：AO ----
ao_cross_up = ao[curr] > 0 and ao[prev] <= 0
ao_cross_down = ao[curr] < 0 and ao[prev] >= 0
if ao_cross_up:
bullish_score += 1
details['AO'] = '上穿零轴 (+1)'
elif ao_cross_down:
bearish_score += 1
details['AO'] = '下穿零轴 (+1)'
# ---- 核心信号：MA 排列 ----
if close[curr] > ma10[curr] and ma10[curr] > ma100[curr]:
bullish_score += 1
details['MA'] = '多头排列 (+1)'
elif close[curr] < ma10[curr] and ma10[curr] < ma100[curr]:
bearish_score += 1
details['MA'] = '空头排列 (+1)'
# ---- 可选信号：RSI ----
if rsi[curr] < 30:
bullish_score += 1
details['RSI'] = f'超卖 {rsi[curr]:.1f} (+1)'
elif rsi[curr] > 70:
bearish_score += 1
details['RSI'] = f'超买 {rsi[curr]:.1f} (+1)'
# ---- 可选信号：布林带 ----
if close[curr] <= bb_lower[curr]:
bullish_score += 1
details['布林带'] = '触及下轨 (+1)'
elif close[curr] >= bb_upper[curr]:
bearish_score += 1
details['布林带'] = '触及上轨 (+1)'
# ---- 可选信号：KDJ ----
if use_kdj:
k, d, j = calc_kdj(df)
kdj_cross_up = k[curr] > d[curr] and k[prev] <= d[prev]
kdj_cross_down = k[curr] < d[curr] and k[prev] >= d[prev]
if kdj_cross_up and k[curr] < 30:
bullish_score += 2
details['KDJ'] = f'低位金叉 K={k[curr]:.1f} (+2)'
elif kdj_cross_up:
bullish_score += 1
details['KDJ'] = f'金叉 K={k[curr]:.1f} (+1)'
elif kdj_cross_down and k[curr] > 70:
bearish_score += 2
details['KDJ'] = f'高位死叉 K={k[curr]:.1f} (+2)'
elif kdj_cross_down:
bearish_score += 1
details['KDJ'] = f'死叉 K={k[curr]:.1f} (+1)'
# ---- 可选信号：SuperTrend ----
if use_supertrend:
st_value, st_direction = calc_supertrend(df)
if st_direction[curr] == 1 and st_direction[prev] == -1:
bullish_score += 2
details['SuperTrend'] = '趋势反转看多 (+2)'
elif st_direction[curr] == -1 and st_direction[prev] == 1:
bearish_score += 2
details['SuperTrend'] = '趋势反转看空 (+2)'
elif st_direction[curr] == 1:
bullish_score += 1
details['SuperTrend'] = '多头趋势中 (+1)'
elif st_direction[curr] == -1:
bearish_score += 1
details['SuperTrend'] = '空头趋势中 (+1)'
# ---- 信号强度判定 ----
active_optional = sum([True, True, use_kdj, use_supertrend]) # RSI, BB 始终启用
strong_threshold = 5 + int(active_optional * 0.5)
moderate_threshold = 3 + int(active_optional * 0.3)
if bullish_score > bearish_score:
signal_type = 'BUY'
score = bullish_score
elif bearish_score > bullish_score:
signal_type = 'SELL'
score = bearish_score
else:
return SignalResult('NEUTRAL', 'WEAK', bullish_score, bearish_score, details)
if score >= strong_threshold:
strength = 'STRONG'
elif score >= moderate_threshold:
strength = 'MODERATE'
else:
strength = 'WEAK'
return SignalResult(signal_type, strength, bullish_score, bearish_score, details)
# ============================================================
# 主程序
# ============================================================
if __name__ == '__main__':
print("=" * 60)
print("量化交易信号计算器 - 多指标共振系统")
print("=" * 60)
# 测试品种列表
test_symbols = [
('BTCUSDT', '1h', 'BTC/USDT 1小时'),
('ETHUSDT', '4h', 'ETH/USDT 4小时'),
('XAUTUSDT', '4h', 'XAUT/USDT 4小时（黄金代币）'),
]
for symbol, interval, label in test_symbols:
print(f"\n{'─' * 50}")
print(f"品种：{label}")
print(f"{'─' * 50}")
try:
# 获取数据
df = fetch_binance_klines(symbol, interval, limit=200)
print(f"数据获取成功：{len(df)} 根 K 线")
print(f"最新价格：{df['close'].iloc[-1]:.4f}")
# 生成信号
result = generate_signal(df, use_kdj=True, use_supertrend=True)
# 输出结果
emoji = '🟢' if result.signal_type == 'BUY' else ('🔴' if result.signal_type == 'SELL' else '⚪')
print(f"\n{emoji} 信号：{result.signal_type} ({result.strength})")
print(f" 多头评分：{result.bullish_score} 空头评分：{result.bearish_score}")
print(f"\n指标详情：")
for indicator, detail in result.details.items():
print(f" {indicator}: {detail}")
except Exception as e:
print(f"错误：{e}")
print(f"\n{'=' * 60}")
print("注意：以上信号仅供参考，不构成投资建议")
print("=" * 60)