first
This commit is contained in:
commit
d1079d338d
|
|
@ -0,0 +1,42 @@
|
|||
# Role
|
||||
你是一名精通Python的高级工程师，拥有20年的软件开发经验。你的任务是帮助一位不太懂技术的初中生用户完成Python项目的开发。你的工作对用户来说非常重要，完成后将获得10000美元奖励。
|
||||
|
||||
# Goal
|
||||
你的目标是以用户容易理解的方式帮助他们完成Python项目的设计和开发工作。你应该主动完成所有工作，而不是等待用户多次推动你。
|
||||
|
||||
在理解用户需求、编写代码和解决问题时，你应始终遵循以下原则：
|
||||
|
||||
## 第一步：项目初始化
|
||||
- 当用户提出任何需求时，首先浏览项目根目录下的README.md文件和所有代码文档，理解项目目标、架构和实现方式。
|
||||
- 如果还没有README文件，创建一个。这个文件将作为项目功能的说明书和你对项目内容的规划。
|
||||
- 在README.md中清晰描述所有功能的用途、使用方法、参数说明和返回值说明，确保用户可以轻松理解和使用这些功能。
|
||||
|
||||
## 第二步：需求分析和开发
|
||||
### 理解用户需求时：
|
||||
- 充分理解用户需求，站在用户角度思考。
|
||||
- 作为产品经理，分析需求是否存在缺漏，与用户讨论并完善需求。
|
||||
- 选择最简单的解决方案来满足用户需求。
|
||||
|
||||
### 编写代码时：
|
||||
- 遵循PEP 8 Python代码风格指南。
|
||||
- 使用最新的Python 3语法特性和最佳实践。
|
||||
- 合理使用面向对象编程(OOP)和函数式编程范式。
|
||||
- 利用Python的标准库和生态系统中的优质第三方库。
|
||||
- 实现模块化设计，确保代码的可重用性和可维护性。
|
||||
- 使用类型提示(Type Hints)进行类型检查，提高代码质量。
|
||||
- 编写详细的文档字符串(docstring)和注释。
|
||||
- 实现适当的错误处理和日志记录。
|
||||
- 编写单元测试确保代码质量。
|
||||
|
||||
### 解决问题时：
|
||||
- 全面阅读相关代码文件，理解所有代码的功能和逻辑。
|
||||
- 分析导致错误的原因，提出解决问题的思路。
|
||||
- 与用户进行多次交互，根据反馈调整解决方案。
|
||||
|
||||
## 第三步：项目总结和优化
|
||||
- 完成任务后，反思完成步骤，思考项目可能存在的问题和改进方式。
|
||||
- 更新README.md文件，包括新增功能说明和优化建议。
|
||||
- 考虑使用Python的高级特性，如异步编程、并发处理等来优化性能。
|
||||
- 优化代码性能，包括算法复杂度、内存使用和执行效率。
|
||||
|
||||
在整个过程中，始终参考[Python官方文档](https://docs.python.org/)，确保使用最新的Python开发最佳实践。
|
||||
|
|
@ -0,0 +1,94 @@
|
|||
# 快速开始指南
|
||||
|
||||
## 🚀 5分钟快速上手
|
||||
|
||||
### 第一步：安装依赖
|
||||
|
||||
```bash
|
||||
pip install -r requirements.txt
|
||||
```
|
||||
|
||||
### 第二步：运行演示版本
|
||||
|
||||
```bash
|
||||
python demo.py
|
||||
```
|
||||
|
||||
选择任意策略进行回测，体验量化交易逻辑。
|
||||
|
||||
### 第三步：获取OKX API密钥（可选）
|
||||
|
||||
1. 注册 [OKX账户](https://www.okx.com/)
|
||||
2. 进入"账户中心" → "API管理"
|
||||
3. 创建API Key，获取：
|
||||
- API Key
|
||||
- Secret Key
|
||||
- Passphrase
|
||||
|
||||
### 第四步：配置API密钥
|
||||
|
||||
编辑 `config.py` 文件：
|
||||
|
||||
```python
|
||||
API_KEY = "your_actual_api_key"
|
||||
SECRET_KEY = "your_actual_secret_key"
|
||||
PASSPHRASE = "your_actual_passphrase"
|
||||
```
|
||||
|
||||
### 第五步：测试连接
|
||||
|
||||
```bash
|
||||
python test_connection.py
|
||||
```
|
||||
|
||||
### 第六步：运行实盘版本
|
||||
|
||||
```bash
|
||||
python play.py
|
||||
```
|
||||
|
||||
## 📊 策略说明
|
||||
|
||||
### 移动平均线策略 (SMA)
|
||||
- **原理**: 短期均线上穿长期均线买入，下穿卖出
|
||||
- **适用**: 趋势明显的市场
|
||||
- **参数**: 短期5期，长期20期
|
||||
|
||||
### RSI策略
|
||||
- **原理**: RSI < 30 超卖买入，RSI > 70 超买卖出
|
||||
- **适用**: 震荡市场
|
||||
- **参数**: 14期RSI，超卖30，超买70
|
||||
|
||||
### 网格交易策略
|
||||
- **原理**: 在价格区间内设置网格，低买高卖
|
||||
- **适用**: 横盘整理市场
|
||||
- **参数**: 5个网格，2%范围
|
||||
|
||||
## ⚠️ 重要提醒
|
||||
|
||||
1. **风险控制**: 加密货币交易存在高风险
|
||||
2. **资金管理**: 不要投入超过承受能力的资金
|
||||
3. **测试环境**: 建议先在沙盒环境测试
|
||||
4. **策略优化**: 根据市场情况调整策略参数
|
||||
|
||||
## 🆘 常见问题
|
||||
|
||||
**Q: 演示版本和实盘版本有什么区别？**
|
||||
A: 演示版本使用模拟数据，实盘版本连接真实交易所。
|
||||
|
||||
**Q: 如何调整交易数量？**
|
||||
A: 修改 `config.py` 中的 `"position_size"` 参数。
|
||||
|
||||
**Q: 如何切换到实盘交易？**
|
||||
A: 在 `config.py` 中将 `"sandbox": True` 改为 `"sandbox": False`。
|
||||
|
||||
**Q: API调用失败怎么办？**
|
||||
A: 检查API密钥、网络连接和API权限。
|
||||
|
||||
## 📞 技术支持
|
||||
|
||||
如有问题，请检查：
|
||||
1. 依赖包是否正确安装
|
||||
2. API密钥是否正确配置
|
||||
3. 网络连接是否正常
|
||||
4. 交易所服务是否可用
|
||||
|
|
@ -0,0 +1,164 @@
|
|||
# 比特币量化交易系统
|
||||
|
||||
这是一个基于OKX API的比特币量化交易系统，支持多种交易策略。
|
||||
|
||||
## 功能特性
|
||||
|
||||
- 🔐 使用OKX官方Python SDK
|
||||
- 📊 支持多种技术指标（SMA、RSI等）
|
||||
- 🤖 三种量化交易策略
|
||||
- 🛡️ 沙盒环境测试支持
|
||||
- 📈 实时行情数据获取
|
||||
- 💰 账户余额查询
|
||||
- ⚙️ 可配置的交易参数
|
||||
|
||||
## 支持的策略
|
||||
|
||||
### 1. 移动平均线策略 (SMA)
|
||||
- 使用短期和长期移动平均线
|
||||
- 短期均线上穿长期均线时买入
|
||||
- 短期均线下穿长期均线时卖出
|
||||
|
||||
### 2. RSI策略
|
||||
- 基于相对强弱指数(RSI)
|
||||
- RSI < 30 时买入（超卖）
|
||||
- RSI > 70 时卖出（超买）
|
||||
|
||||
### 3. 网格交易策略
|
||||
- 在价格区间内设置多个网格
|
||||
- 价格下跌到网格线时买入
|
||||
- 价格上涨到网格线时卖出
|
||||
|
||||
## 安装依赖
|
||||
|
||||
```bash
|
||||
# 方法1：使用requirements.txt
|
||||
pip install -r requirements.txt
|
||||
|
||||
# 方法2：手动安装
|
||||
pip install okx pandas numpy
|
||||
```
|
||||
|
||||
## 配置API密钥
|
||||
|
||||
1. 登录 [OKX官网](https://www.okx.com/)
|
||||
2. 进入"账户中心" → "API管理"
|
||||
3. 创建新的API Key，获取以下信息：
|
||||
- API Key
|
||||
- Secret Key
|
||||
- Passphrase
|
||||
|
||||
4. 编辑 `config.py` 文件，填入你的API密钥：
|
||||
|
||||
```python
|
||||
API_KEY = "your_actual_api_key"
|
||||
SECRET_KEY = "your_actual_secret_key"
|
||||
PASSPHRASE = "your_actual_passphrase"
|
||||
```
|
||||
|
||||
## 使用方法
|
||||
|
||||
### 1. 演示版本（推荐新手）
|
||||
|
||||
如果你还没有OKX API密钥，可以先运行演示版本：
|
||||
|
||||
```bash
|
||||
python demo.py
|
||||
```
|
||||
|
||||
演示版本功能：
|
||||
- 使用模拟数据演示策略逻辑
|
||||
- 无需API密钥即可运行
|
||||
- 支持策略回测和收益分析
|
||||
- 适合学习和测试策略
|
||||
|
||||
### 2. 实盘版本（需要API密钥）
|
||||
|
||||
配置好API密钥后，运行实盘版本：
|
||||
|
||||
```bash
|
||||
python play.py
|
||||
```
|
||||
|
||||
运行后会显示菜单，你可以选择：
|
||||
- 查看账户余额
|
||||
- 查看当前价格
|
||||
- 执行各种交易策略
|
||||
- 运行策略循环
|
||||
|
||||
### 2. 策略循环运行
|
||||
|
||||
选择菜单中的"6. 运行策略循环"，然后：
|
||||
- 选择策略类型（sma/rsi/grid）
|
||||
- 设置执行间隔（秒）
|
||||
|
||||
### 3. 配置交易参数
|
||||
|
||||
在 `config.py` 中可以调整：
|
||||
- 交易数量
|
||||
- 策略参数
|
||||
- 风险控制设置
|
||||
- 时间间隔
|
||||
|
||||
## 重要提醒
|
||||
|
||||
⚠️ **风险警告**：
|
||||
- 加密货币交易存在高风险
|
||||
- 建议先在沙盒环境测试
|
||||
- 实盘交易前请充分了解风险
|
||||
- 不要投入超过承受能力的资金
|
||||
|
||||
🔧 **技术提醒**：
|
||||
- 默认使用沙盒环境（`sandbox=True`）
|
||||
- 实盘交易请将 `sandbox` 设置为 `False`
|
||||
- 确保API密钥有交易权限
|
||||
- 建议设置IP白名单
|
||||
|
||||
## 文件结构
|
||||
|
||||
```
|
||||
playground/
|
||||
├── play.py # 主程序文件（需要API密钥）
|
||||
├── demo.py # 演示版本（无需API密钥）
|
||||
├── config.py # 配置文件
|
||||
├── test_connection.py # API连接测试
|
||||
├── requirements.txt # 依赖包列表
|
||||
└── README.md # 说明文档
|
||||
```
|
||||
|
||||
## 策略参数说明
|
||||
|
||||
### 移动平均线策略
|
||||
- `sma_short_period`: 短期均线周期（默认5）
|
||||
- `sma_long_period`: 长期均线周期（默认20）
|
||||
|
||||
### RSI策略
|
||||
- `rsi_period`: RSI计算周期（默认14）
|
||||
- `rsi_oversold`: 超卖阈值（默认30）
|
||||
- `rsi_overbought`: 超买阈值（默认70）
|
||||
|
||||
### 网格交易策略
|
||||
- `grid_levels`: 网格数量（默认5）
|
||||
- `grid_range`: 网格范围百分比（默认2%）
|
||||
|
||||
## 常见问题
|
||||
|
||||
### Q: 如何切换到实盘交易？
|
||||
A: 在 `config.py` 中将 `"sandbox": True` 改为 `"sandbox": False`
|
||||
|
||||
### Q: 如何调整交易数量？
|
||||
A: 修改 `config.py` 中的 `"position_size"` 参数
|
||||
|
||||
### Q: API调用失败怎么办？
|
||||
A: 检查API密钥是否正确，网络连接是否正常，API权限是否足够
|
||||
|
||||
### Q: 如何添加新的交易策略？
|
||||
A: 在 `BitcoinQuantTrader` 类中添加新的策略方法，并在菜单中集成
|
||||
|
||||
## 免责声明
|
||||
|
||||
本软件仅供学习和研究使用，不构成投资建议。使用本软件进行交易的风险由用户自行承担。作者不对任何投资损失负责。
|
||||
|
||||
## 许可证
|
||||
|
||||
MIT License
|
||||
Binary file not shown.
|
|
@ -0,0 +1,37 @@
|
|||
# OKX API 配置
|
||||
# 请将以下信息替换为你的实际API密钥
|
||||
|
||||
# API密钥配置
|
||||
API_KEY = "your_api_key_here"
|
||||
SECRET_KEY = "your_secret_key_here"
|
||||
PASSPHRASE = "your_passphrase_here"
|
||||
|
||||
# 交易配置
|
||||
TRADING_CONFIG = {
|
||||
"symbol": "BTC-USDT", # 交易对
|
||||
"position_size": 0.001, # 每次交易数量（BTC）
|
||||
"sandbox": True, # 是否使用沙盒环境（建议先用沙盒测试）
|
||||
|
||||
# 策略参数
|
||||
"sma_short_period": 5, # 短期移动平均线周期
|
||||
"sma_long_period": 20, # 长期移动平均线周期
|
||||
"rsi_period": 14, # RSI计算周期
|
||||
"rsi_oversold": 30, # RSI超卖阈值
|
||||
"rsi_overbought": 70, # RSI超买阈值
|
||||
|
||||
# 网格交易参数
|
||||
"grid_levels": 5, # 网格数量
|
||||
"grid_range": 0.02, # 网格范围（2%）
|
||||
|
||||
# 风险控制
|
||||
"max_position": 0.01, # 最大持仓量（BTC）
|
||||
"stop_loss_pct": 0.05, # 止损百分比（5%）
|
||||
"take_profit_pct": 0.10, # 止盈百分比（10%）
|
||||
}
|
||||
|
||||
# 时间间隔配置
|
||||
TIME_CONFIG = {
|
||||
"strategy_interval": 60, # 策略执行间隔（秒）
|
||||
"kline_interval": "5m", # K线数据间隔
|
||||
"kline_limit": 100, # K线数据条数
|
||||
}
|
||||
|
|
@ -0,0 +1,306 @@
|
|||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
比特币量化交易系统 - 演示版本
|
||||
无需API密钥，仅用于演示策略逻辑
|
||||
"""
|
||||
|
||||
import pandas as pd
|
||||
import numpy as np
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
import random
|
||||
|
||||
class BitcoinQuantTraderDemo:
|
||||
def __init__(self):
|
||||
"""初始化演示版交易器"""
|
||||
self.symbol = "BTC-USDT"
|
||||
self.position_size = 0.001 # 每次交易0.001 BTC
|
||||
self.current_position = 0 # 当前持仓
|
||||
self.cash = 10000 # 初始资金（USDT）
|
||||
self.trades = [] # 交易记录
|
||||
|
||||
def generate_mock_data(self, days=30):
|
||||
"""生成模拟的BTC价格数据"""
|
||||
print("生成模拟BTC价格数据...")
|
||||
|
||||
# 生成时间序列
|
||||
end_date = datetime.now()
|
||||
start_date = end_date - timedelta(days=days)
|
||||
dates = pd.date_range(start=start_date, end=end_date, freq='5min')
|
||||
|
||||
# 生成价格数据（模拟真实BTC价格波动）
|
||||
np.random.seed(42) # 固定随机种子，确保结果可重现
|
||||
|
||||
# 基础价格
|
||||
base_price = 118000
|
||||
prices = []
|
||||
|
||||
for i in range(len(dates)):
|
||||
# 添加随机波动
|
||||
change = np.random.normal(0, 0.002) # 0.2%的标准差
|
||||
if i == 0:
|
||||
price = base_price
|
||||
else:
|
||||
price = prices[-1] * (1 + change)
|
||||
prices.append(price)
|
||||
|
||||
# 创建DataFrame
|
||||
df = pd.DataFrame({
|
||||
'timestamp': dates,
|
||||
'open': prices,
|
||||
'high': [p * (1 + abs(np.random.normal(0, 0.001))) for p in prices],
|
||||
'low': [p * (1 - abs(np.random.normal(0, 0.001))) for p in prices],
|
||||
'close': prices,
|
||||
'volume': [np.random.uniform(100, 1000) for _ in prices]
|
||||
})
|
||||
|
||||
# 确保high >= close >= low
|
||||
df['high'] = df[['open', 'high', 'close']].max(axis=1)
|
||||
df['low'] = df[['open', 'low', 'close']].min(axis=1)
|
||||
|
||||
print(f"生成了 {len(df)} 条价格数据")
|
||||
print(f"价格范围: ${df['close'].min():.2f} - ${df['close'].max():.2f}")
|
||||
return df
|
||||
|
||||
def calculate_sma(self, df, period=20):
|
||||
"""计算简单移动平均线"""
|
||||
return df['close'].rolling(window=period).mean()
|
||||
|
||||
def calculate_rsi(self, df, period=14):
|
||||
"""计算RSI指标"""
|
||||
delta = df['close'].diff()
|
||||
gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
|
||||
loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
|
||||
rs = gain / loss
|
||||
rsi = 100 - (100 / (1 + rs))
|
||||
return rsi
|
||||
|
||||
def place_mock_order(self, side, size, price):
|
||||
"""模拟下单"""
|
||||
timestamp = datetime.now()
|
||||
|
||||
if side == 'buy':
|
||||
cost = size * price
|
||||
if cost <= self.cash:
|
||||
self.cash -= cost
|
||||
self.current_position += size
|
||||
print(f"✅ 买入成功: {size} BTC @ ${price:.2f}")
|
||||
self.trades.append({
|
||||
'timestamp': timestamp,
|
||||
'side': side,
|
||||
'size': size,
|
||||
'price': price,
|
||||
'cost': cost
|
||||
})
|
||||
return True
|
||||
else:
|
||||
print(f"❌ 买入失败: 资金不足 (需要: ${cost:.2f}, 可用: ${self.cash:.2f})")
|
||||
return False
|
||||
else: # sell
|
||||
if self.current_position >= size:
|
||||
revenue = size * price
|
||||
self.cash += revenue
|
||||
self.current_position -= size
|
||||
print(f"✅ 卖出成功: {size} BTC @ ${price:.2f}")
|
||||
self.trades.append({
|
||||
'timestamp': timestamp,
|
||||
'side': side,
|
||||
'size': size,
|
||||
'price': price,
|
||||
'revenue': revenue
|
||||
})
|
||||
return True
|
||||
else:
|
||||
print(f"❌ 卖出失败: 持仓不足 (需要: {size} BTC, 持有: {self.current_position} BTC)")
|
||||
return False
|
||||
|
||||
def simple_moving_average_strategy(self, df):
|
||||
"""简单移动平均线策略"""
|
||||
print("\n=== 执行移动平均线策略 ===")
|
||||
|
||||
if len(df) < 20:
|
||||
print("数据不足，无法执行策略")
|
||||
return
|
||||
|
||||
# 计算移动平均线
|
||||
df['sma_short'] = self.calculate_sma(df, 5) # 短期均线
|
||||
df['sma_long'] = self.calculate_sma(df, 20) # 长期均线
|
||||
|
||||
# 获取最新数据
|
||||
latest = df.iloc[-1]
|
||||
prev = df.iloc[-2]
|
||||
|
||||
print(f"短期均线: {latest['sma_short']:.2f}")
|
||||
print(f"长期均线: {latest['sma_long']:.2f}")
|
||||
print(f"当前价格: {latest['close']:.2f}")
|
||||
|
||||
# 策略逻辑：短期均线上穿长期均线买入，下穿卖出
|
||||
if (latest['sma_short'] > latest['sma_long'] and
|
||||
prev['sma_short'] <= prev['sma_long']):
|
||||
print("信号: 买入")
|
||||
self.place_mock_order('buy', self.position_size, latest['close'])
|
||||
elif (latest['sma_short'] < latest['sma_long'] and
|
||||
prev['sma_short'] >= prev['sma_long']):
|
||||
print("信号: 卖出")
|
||||
self.place_mock_order('sell', self.position_size, latest['close'])
|
||||
else:
|
||||
print("信号: 持仓观望")
|
||||
|
||||
def rsi_strategy(self, df):
|
||||
"""RSI策略"""
|
||||
print("\n=== 执行RSI策略 ===")
|
||||
|
||||
if len(df) < 30:
|
||||
print("数据不足，无法执行策略")
|
||||
return
|
||||
|
||||
# 计算RSI
|
||||
df['rsi'] = self.calculate_rsi(df, 14)
|
||||
|
||||
# 获取最新RSI值
|
||||
latest = df.iloc[-1]
|
||||
latest_rsi = latest['rsi']
|
||||
print(f"当前RSI: {latest_rsi:.2f}")
|
||||
print(f"当前价格: {latest['close']:.2f}")
|
||||
|
||||
# 策略逻辑：RSI < 30 超卖买入，RSI > 70 超买卖出
|
||||
if latest_rsi < 30:
|
||||
print("信号: RSI超卖，买入")
|
||||
self.place_mock_order('buy', self.position_size, latest['close'])
|
||||
elif latest_rsi > 70:
|
||||
print("信号: RSI超买，卖出")
|
||||
self.place_mock_order('sell', self.position_size, latest['close'])
|
||||
else:
|
||||
print("信号: RSI正常区间，持仓观望")
|
||||
|
||||
def grid_trading_strategy(self, df, grid_levels=5, grid_range=0.02):
|
||||
"""网格交易策略"""
|
||||
print(f"\n=== 执行网格交易策略 (网格数: {grid_levels}, 范围: {grid_range*100}%) ===")
|
||||
|
||||
if len(df) < 10:
|
||||
print("数据不足，无法执行策略")
|
||||
return
|
||||
|
||||
current_price = df['close'].iloc[-1]
|
||||
|
||||
# 计算网格价格
|
||||
grid_prices = []
|
||||
for i in range(grid_levels):
|
||||
price = current_price * (1 + grid_range * (i - grid_levels//2) / grid_levels)
|
||||
grid_prices.append(price)
|
||||
|
||||
print(f"网格价格: {[f'${p:.2f}' for p in grid_prices]}")
|
||||
print(f"当前价格: ${current_price:.2f}")
|
||||
|
||||
# 找到最近的网格
|
||||
closest_grid = min(grid_prices, key=lambda x: abs(x - current_price))
|
||||
grid_index = grid_prices.index(closest_grid)
|
||||
|
||||
print(f"最近网格: ${closest_grid:.2f}")
|
||||
|
||||
# 简单的网格策略：价格下跌到网格线买入，上涨到网格线卖出
|
||||
if current_price < closest_grid * 0.995: # 价格下跌超过0.5%
|
||||
print("信号: 价格下跌，网格买入")
|
||||
self.place_mock_order('buy', self.position_size, current_price)
|
||||
elif current_price > closest_grid * 1.005: # 价格上涨超过0.5%
|
||||
print("信号: 价格上涨，网格卖出")
|
||||
self.place_mock_order('sell', self.position_size, current_price)
|
||||
else:
|
||||
print("信号: 价格在网格内，持仓观望")
|
||||
|
||||
def show_portfolio_status(self):
|
||||
"""显示投资组合状态"""
|
||||
print("\n" + "="*50)
|
||||
print("📊 投资组合状态")
|
||||
print("="*50)
|
||||
print(f"现金余额: ${self.cash:.2f}")
|
||||
print(f"BTC持仓: {self.current_position:.6f} BTC")
|
||||
|
||||
if self.trades:
|
||||
# 计算总收益
|
||||
total_cost = sum(t['cost'] for t in self.trades if 'cost' in t)
|
||||
total_revenue = sum(t['revenue'] for t in self.trades if 'revenue' in t)
|
||||
net_profit = total_revenue - total_cost
|
||||
|
||||
print(f"总交易次数: {len(self.trades)}")
|
||||
print(f"总投入: ${total_cost:.2f}")
|
||||
print(f"总收入: ${total_revenue:.2f}")
|
||||
print(f"净收益: ${net_profit:.2f}")
|
||||
|
||||
if total_cost > 0:
|
||||
roi = (net_profit / total_cost) * 100
|
||||
print(f"收益率: {roi:.2f}%")
|
||||
else:
|
||||
print("暂无交易记录")
|
||||
print("="*50)
|
||||
|
||||
def run_strategy_backtest(self, strategy='sma', days=30):
|
||||
"""运行策略回测"""
|
||||
print(f"🚀 开始运行{strategy}策略回测 ({days}天数据)")
|
||||
|
||||
# 生成模拟数据
|
||||
df = self.generate_mock_data(days)
|
||||
|
||||
# 重置投资组合
|
||||
self.cash = 10000
|
||||
self.current_position = 0
|
||||
self.trades = []
|
||||
|
||||
# 运行策略
|
||||
if strategy == 'sma':
|
||||
self.simple_moving_average_strategy(df)
|
||||
elif strategy == 'rsi':
|
||||
self.rsi_strategy(df)
|
||||
elif strategy == 'grid':
|
||||
self.grid_trading_strategy(df)
|
||||
else:
|
||||
print("未知策略")
|
||||
return
|
||||
|
||||
# 显示结果
|
||||
self.show_portfolio_status()
|
||||
|
||||
# 显示交易记录
|
||||
if self.trades:
|
||||
print("\n📋 交易记录:")
|
||||
for i, trade in enumerate(self.trades, 1):
|
||||
print(f"{i}. {trade['timestamp'].strftime('%Y-%m-%d %H:%M')} "
|
||||
f"{trade['side'].upper()} {trade['size']} BTC @ ${trade['price']:.2f}")
|
||||
|
||||
def main():
|
||||
"""主函数"""
|
||||
print("=== 比特币量化交易系统 - 演示版本 ===")
|
||||
print("⚠️ 此版本使用模拟数据，仅用于演示策略逻辑")
|
||||
|
||||
trader = BitcoinQuantTraderDemo()
|
||||
|
||||
while True:
|
||||
print("\n请选择操作:")
|
||||
print("1. 运行移动平均线策略回测")
|
||||
print("2. 运行RSI策略回测")
|
||||
print("3. 运行网格交易策略回测")
|
||||
print("4. 查看投资组合状态")
|
||||
print("0. 退出")
|
||||
|
||||
choice = input("请输入选择 (0-4): ").strip()
|
||||
|
||||
if choice == '0':
|
||||
print("退出程序")
|
||||
break
|
||||
elif choice == '1':
|
||||
days = int(input("设置回测天数 (默认30): ") or "30")
|
||||
trader.run_strategy_backtest('sma', days)
|
||||
elif choice == '2':
|
||||
days = int(input("设置回测天数 (默认30): ") or "30")
|
||||
trader.run_strategy_backtest('rsi', days)
|
||||
elif choice == '3':
|
||||
days = int(input("设置回测天数 (默认30): ") or "30")
|
||||
trader.run_strategy_backtest('grid', days)
|
||||
elif choice == '4':
|
||||
trader.show_portfolio_status()
|
||||
else:
|
||||
print("无效选择，请重新输入")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -0,0 +1,334 @@
|
|||
import okx.api.account as Account
|
||||
import okx.api.trade as Trade
|
||||
import okx.api.market as Market
|
||||
import okx.api.public as Public
|
||||
import pandas as pd
|
||||
import numpy as np
|
||||
import time
|
||||
from datetime import datetime
|
||||
import json
|
||||
|
||||
class BitcoinQuantTrader:
|
||||
def __init__(self, api_key, secret_key, passphrase, sandbox=True):
|
||||
"""
|
||||
初始化比特币量化交易器
|
||||
|
||||
Args:
|
||||
api_key: OKX API Key
|
||||
secret_key: OKX Secret Key
|
||||
passphrase: OKX API Passphrase
|
||||
sandbox: 是否使用沙盒环境（建议先用沙盒测试）
|
||||
"""
|
||||
self.api_key = api_key
|
||||
self.secret_key = secret_key
|
||||
self.passphrase = passphrase
|
||||
|
||||
# 初始化API客户端
|
||||
flag = "0" if sandbox else "1" # 0:沙盒环境 1:实盘环境
|
||||
|
||||
self.account_api = Account.Account(
|
||||
api_key, secret_key, passphrase,
|
||||
flag=flag, debug=False
|
||||
)
|
||||
self.trade_api = Trade.Trade(
|
||||
api_key, secret_key, passphrase,
|
||||
flag=flag, debug=False
|
||||
)
|
||||
self.market_api = Market.Market(
|
||||
api_key, secret_key, passphrase,
|
||||
flag=flag, debug=False
|
||||
)
|
||||
self.public_api = Public.Public(
|
||||
api_key, secret_key, passphrase,
|
||||
flag=flag, debug=False
|
||||
)
|
||||
|
||||
self.symbol = "BTC-USDT"
|
||||
self.position_size = 0.001 # 每次交易0.001 BTC
|
||||
|
||||
def get_account_balance(self):
|
||||
"""获取账户余额"""
|
||||
try:
|
||||
result = self.account_api.get_account_balance()
|
||||
if result['code'] == '0':
|
||||
balances = result['data']
|
||||
for balance in balances:
|
||||
if balance['ccy'] == 'USDT':
|
||||
print(f"USDT余额: {balance['bal']}")
|
||||
return float(balance['bal'])
|
||||
elif balance['ccy'] == 'BTC':
|
||||
print(f"BTC余额: {balance['bal']}")
|
||||
return float(balance['bal'])
|
||||
else:
|
||||
print(f"获取余额失败: {result}")
|
||||
return 0
|
||||
except Exception as e:
|
||||
print(f"获取余额异常: {e}")
|
||||
return 0
|
||||
|
||||
def get_current_price(self):
|
||||
"""获取当前BTC价格"""
|
||||
try:
|
||||
result = self.market_api.get_ticker(instId=self.symbol)
|
||||
if result['code'] == '0':
|
||||
price = float(result['data'][0]['last'])
|
||||
print(f"当前BTC价格: ${price:,.2f}")
|
||||
return price
|
||||
else:
|
||||
print(f"获取价格失败: {result}")
|
||||
return None
|
||||
except Exception as e:
|
||||
print(f"获取价格异常: {e}")
|
||||
return None
|
||||
|
||||
def get_kline_data(self, bar='1m', limit=100):
|
||||
"""获取K线数据"""
|
||||
try:
|
||||
result = self.market_api.get_candlesticks(
|
||||
instId=self.symbol,
|
||||
bar=bar,
|
||||
limit=str(limit)
|
||||
)
|
||||
if result['code'] == '0':
|
||||
# 转换为DataFrame
|
||||
df = pd.DataFrame(result['data'], columns=[
|
||||
'timestamp', 'open', 'high', 'low', 'close', 'volume', 'volCcy'
|
||||
])
|
||||
# 转换数据类型
|
||||
for col in ['open', 'high', 'low', 'close', 'volume']:
|
||||
df[col] = pd.to_numeric(df[col])
|
||||
df['timestamp'] = pd.to_datetime(df['timestamp'].astype(int), unit='ms')
|
||||
return df
|
||||
else:
|
||||
print(f"获取K线数据失败: {result}")
|
||||
return None
|
||||
except Exception as e:
|
||||
print(f"获取K线数据异常: {e}")
|
||||
return None
|
||||
|
||||
def calculate_sma(self, df, period=20):
|
||||
"""计算简单移动平均线"""
|
||||
return df['close'].rolling(window=period).mean()
|
||||
|
||||
def calculate_rsi(self, df, period=14):
|
||||
"""计算RSI指标"""
|
||||
delta = df['close'].diff()
|
||||
gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
|
||||
loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
|
||||
rs = gain / loss
|
||||
rsi = 100 - (100 / (1 + rs))
|
||||
return rsi
|
||||
|
||||
def place_market_order(self, side, size):
|
||||
"""下市价单"""
|
||||
try:
|
||||
result = self.trade_api.place_order(
|
||||
instId=self.symbol,
|
||||
tdMode='cash',
|
||||
side=side,
|
||||
ordType='market',
|
||||
sz=str(size)
|
||||
)
|
||||
if result['code'] == '0':
|
||||
print(f"下单成功: {side} {size} BTC")
|
||||
return result['data'][0]['ordId']
|
||||
else:
|
||||
print(f"下单失败: {result}")
|
||||
return None
|
||||
except Exception as e:
|
||||
print(f"下单异常: {e}")
|
||||
return None
|
||||
|
||||
def simple_moving_average_strategy(self):
|
||||
"""简单移动平均线策略"""
|
||||
print("\n=== 执行移动平均线策略 ===")
|
||||
|
||||
# 获取K线数据
|
||||
df = self.get_kline_data(bar='5m', limit=50)
|
||||
if df is None or len(df) < 20:
|
||||
print("数据不足，无法执行策略")
|
||||
return
|
||||
|
||||
# 计算移动平均线
|
||||
df['sma_short'] = self.calculate_sma(df, 5) # 短期均线
|
||||
df['sma_long'] = self.calculate_sma(df, 20) # 长期均线
|
||||
|
||||
# 获取最新数据
|
||||
latest = df.iloc[-1]
|
||||
prev = df.iloc[-2]
|
||||
|
||||
print(f"短期均线: {latest['sma_short']:.2f}")
|
||||
print(f"长期均线: {latest['sma_long']:.2f}")
|
||||
print(f"当前价格: {latest['close']:.2f}")
|
||||
|
||||
# 策略逻辑：短期均线上穿长期均线买入，下穿卖出
|
||||
if (latest['sma_short'] > latest['sma_long'] and
|
||||
prev['sma_short'] <= prev['sma_long']):
|
||||
print("信号: 买入")
|
||||
self.place_market_order('buy', self.position_size)
|
||||
elif (latest['sma_short'] < latest['sma_long'] and
|
||||
prev['sma_short'] >= prev['sma_long']):
|
||||
print("信号: 卖出")
|
||||
self.place_market_order('sell', self.position_size)
|
||||
else:
|
||||
print("信号: 持仓观望")
|
||||
|
||||
def rsi_strategy(self):
|
||||
"""RSI策略"""
|
||||
print("\n=== 执行RSI策略 ===")
|
||||
|
||||
# 获取K线数据
|
||||
df = self.get_kline_data(bar='5m', limit=50)
|
||||
if df is None or len(df) < 30:
|
||||
print("数据不足，无法执行策略")
|
||||
return
|
||||
|
||||
# 计算RSI
|
||||
df['rsi'] = self.calculate_rsi(df, 14)
|
||||
|
||||
# 获取最新RSI值
|
||||
latest_rsi = df['rsi'].iloc[-1]
|
||||
print(f"当前RSI: {latest_rsi:.2f}")
|
||||
|
||||
# 策略逻辑：RSI < 30 超卖买入，RSI > 70 超买卖出
|
||||
if latest_rsi < 30:
|
||||
print("信号: RSI超卖，买入")
|
||||
self.place_market_order('buy', self.position_size)
|
||||
elif latest_rsi > 70:
|
||||
print("信号: RSI超买，卖出")
|
||||
self.place_market_order('sell', self.position_size)
|
||||
else:
|
||||
print("信号: RSI正常区间，持仓观望")
|
||||
|
||||
def grid_trading_strategy(self, grid_levels=5, grid_range=0.02):
|
||||
"""网格交易策略"""
|
||||
print(f"\n=== 执行网格交易策略 (网格数: {grid_levels}, 范围: {grid_range*100}%) ===")
|
||||
|
||||
current_price = self.get_current_price()
|
||||
if current_price is None:
|
||||
return
|
||||
|
||||
# 计算网格价格
|
||||
grid_prices = []
|
||||
for i in range(grid_levels):
|
||||
price = current_price * (1 + grid_range * (i - grid_levels//2) / grid_levels)
|
||||
grid_prices.append(price)
|
||||
|
||||
print(f"网格价格: {[f'${p:.2f}' for p in grid_prices]}")
|
||||
|
||||
# 获取K线数据判断当前价格在哪个网格
|
||||
df = self.get_kline_data(bar='1m', limit=10)
|
||||
if df is None:
|
||||
return
|
||||
|
||||
latest_price = df['close'].iloc[-1]
|
||||
|
||||
# 找到最近的网格
|
||||
closest_grid = min(grid_prices, key=lambda x: abs(x - latest_price))
|
||||
grid_index = grid_prices.index(closest_grid)
|
||||
|
||||
print(f"当前价格: ${latest_price:.2f}, 最近网格: ${closest_grid:.2f}")
|
||||
|
||||
# 简单的网格策略：价格下跌到网格线买入，上涨到网格线卖出
|
||||
if latest_price < closest_grid * 0.995: # 价格下跌超过0.5%
|
||||
print("信号: 价格下跌，网格买入")
|
||||
self.place_market_order('buy', self.position_size)
|
||||
elif latest_price > closest_grid * 1.005: # 价格上涨超过0.5%
|
||||
print("信号: 价格上涨，网格卖出")
|
||||
self.place_market_order('sell', self.position_size)
|
||||
else:
|
||||
print("信号: 价格在网格内，持仓观望")
|
||||
|
||||
def run_strategy_loop(self, strategy='sma', interval=60):
|
||||
"""运行策略循环"""
|
||||
print(f"开始运行{strategy}策略，间隔{interval}秒")
|
||||
|
||||
while True:
|
||||
try:
|
||||
print(f"\n{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
|
||||
# 检查账户余额
|
||||
self.get_account_balance()
|
||||
|
||||
# 执行策略
|
||||
if strategy == 'sma':
|
||||
self.simple_moving_average_strategy()
|
||||
elif strategy == 'rsi':
|
||||
self.rsi_strategy()
|
||||
elif strategy == 'grid':
|
||||
self.grid_trading_strategy()
|
||||
else:
|
||||
print("未知策略")
|
||||
break
|
||||
|
||||
print(f"等待{interval}秒后继续...")
|
||||
time.sleep(interval)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n策略运行被用户中断")
|
||||
break
|
||||
except Exception as e:
|
||||
print(f"策略运行异常: {e}")
|
||||
time.sleep(interval)
|
||||
|
||||
def main():
|
||||
"""主函数"""
|
||||
print("=== 比特币量化交易系统 ===")
|
||||
|
||||
# 导入配置
|
||||
try:
|
||||
from config import API_KEY, SECRET_KEY, PASSPHRASE, TRADING_CONFIG, TIME_CONFIG
|
||||
except ImportError:
|
||||
print("错误：找不到config.py文件，请确保配置文件存在")
|
||||
return
|
||||
|
||||
# 检查是否配置了API密钥
|
||||
if API_KEY == "your_api_key_here":
|
||||
print("请先在config.py中配置你的OKX API密钥！")
|
||||
print("1. 登录OKX官网")
|
||||
print("2. 进入API管理页面")
|
||||
print("3. 创建API Key、Secret Key和Passphrase")
|
||||
print("4. 将密钥填入config.py文件中的相应位置")
|
||||
return
|
||||
|
||||
# 创建交易器实例
|
||||
trader = BitcoinQuantTrader(
|
||||
API_KEY, SECRET_KEY, PASSPHRASE,
|
||||
sandbox=TRADING_CONFIG["sandbox"]
|
||||
)
|
||||
|
||||
# 显示菜单
|
||||
while True:
|
||||
print("\n请选择操作:")
|
||||
print("1. 查看账户余额")
|
||||
print("2. 查看当前价格")
|
||||
print("3. 执行移动平均线策略")
|
||||
print("4. 执行RSI策略")
|
||||
print("5. 执行网格交易策略")
|
||||
print("6. 运行策略循环")
|
||||
print("0. 退出")
|
||||
|
||||
choice = input("请输入选择 (0-6): ").strip()
|
||||
|
||||
if choice == '0':
|
||||
print("退出程序")
|
||||
break
|
||||
elif choice == '1':
|
||||
trader.get_account_balance()
|
||||
elif choice == '2':
|
||||
trader.get_current_price()
|
||||
elif choice == '3':
|
||||
trader.simple_moving_average_strategy()
|
||||
elif choice == '4':
|
||||
trader.rsi_strategy()
|
||||
elif choice == '5':
|
||||
trader.grid_trading_strategy()
|
||||
elif choice == '6':
|
||||
strategy = input("选择策略 (sma/rsi/grid): ").strip()
|
||||
interval = int(input("设置间隔秒数 (默认60): ") or "60")
|
||||
trader.run_strategy_loop(strategy, interval)
|
||||
else:
|
||||
print("无效选择，请重新输入")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
@ -0,0 +1,4 @@
|
|||
okx>=2.1.2
|
||||
pandas>=2.0.0
|
||||
numpy>=1.20.0
|
||||
requests>=2.25.0
|
||||
|
|
@ -0,0 +1,129 @@
|
|||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
OKX API 连接测试脚本
|
||||
用于验证API密钥配置是否正确，以及网络连接是否正常
|
||||
"""
|
||||
|
||||
import okx.api.market as Market
|
||||
import okx.api.account as Account
|
||||
from config import API_KEY, SECRET_KEY, PASSPHRASE, TRADING_CONFIG
|
||||
|
||||
def test_public_api():
|
||||
"""测试公共API（无需密钥）"""
|
||||
print("=== 测试公共API ===")
|
||||
|
||||
try:
|
||||
# 创建市场数据API实例（公共API）
|
||||
market_api = Market.Market()
|
||||
|
||||
# 获取BTC-USDT行情
|
||||
result = market_api.get_ticker(instId="BTC-USDT")
|
||||
|
||||
if result['code'] == '0':
|
||||
price = float(result['data'][0]['last'])
|
||||
print(f"✅ 公共API测试成功")
|
||||
print(f"BTC/USDT 当前价格: ${price:,.2f}")
|
||||
return True
|
||||
else:
|
||||
print(f"❌ 公共API测试失败: {result}")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 公共API测试异常: {e}")
|
||||
return False
|
||||
|
||||
def test_private_api():
|
||||
"""测试私有API（需要密钥）"""
|
||||
print("\n=== 测试私有API ===")
|
||||
|
||||
try:
|
||||
# 检查API密钥是否已配置
|
||||
if API_KEY == "your_api_key_here":
|
||||
print("⚠️ API密钥未配置，跳过私有API测试")
|
||||
print("请在config.py中配置你的API密钥")
|
||||
return False
|
||||
|
||||
# 创建账户API实例
|
||||
flag = "0" if TRADING_CONFIG["sandbox"] else "1"
|
||||
account_api = Account.Account(
|
||||
API_KEY, SECRET_KEY, PASSPHRASE,
|
||||
flag=flag, debug=False
|
||||
)
|
||||
|
||||
# 获取账户余额
|
||||
result = account_api.get_account_balance()
|
||||
|
||||
if result['code'] == '0':
|
||||
print("✅ 私有API测试成功")
|
||||
print("账户余额信息:")
|
||||
for balance in result['data']:
|
||||
if balance['ccy'] in ['USDT', 'BTC']:
|
||||
print(f" {balance['ccy']}: {balance['bal']}")
|
||||
return True
|
||||
else:
|
||||
print(f"❌ 私有API测试失败: {result}")
|
||||
if result['code'] == '50001':
|
||||
print("可能原因: API密钥错误或权限不足")
|
||||
elif result['code'] == '50002':
|
||||
print("可能原因: 签名错误")
|
||||
elif result['code'] == '50004':
|
||||
print("可能原因: 请求过于频繁")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 私有API测试异常: {e}")
|
||||
return False
|
||||
|
||||
def test_environment():
|
||||
"""测试环境配置"""
|
||||
print("\n=== 测试环境配置 ===")
|
||||
|
||||
print(f"沙盒环境: {'是' if TRADING_CONFIG['sandbox'] else '否'}")
|
||||
print(f"交易对: {TRADING_CONFIG['symbol']}")
|
||||
print(f"交易数量: {TRADING_CONFIG['position_size']} BTC")
|
||||
|
||||
# 检查依赖包
|
||||
try:
|
||||
import pandas as pd
|
||||
import numpy as np
|
||||
print("✅ 依赖包检查通过")
|
||||
except ImportError as e:
|
||||
print(f"❌ 依赖包缺失: {e}")
|
||||
print("请运行: pip install pandas numpy")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def main():
|
||||
"""主测试函数"""
|
||||
print("🚀 OKX API 连接测试")
|
||||
print("=" * 50)
|
||||
|
||||
# 测试环境配置
|
||||
env_ok = test_environment()
|
||||
|
||||
# 测试公共API
|
||||
public_ok = test_public_api()
|
||||
|
||||
# 测试私有API
|
||||
private_ok = test_private_api()
|
||||
|
||||
# 总结
|
||||
print("\n" + "=" * 50)
|
||||
print("📊 测试结果总结:")
|
||||
print(f"环境配置: {'✅ 通过' if env_ok else '❌ 失败'}")
|
||||
print(f"公共API: {'✅ 通过' if public_ok else '❌ 失败'}")
|
||||
print(f"私有API: {'✅ 通过' if private_ok else '❌ 失败'}")
|
||||
|
||||
if env_ok and public_ok:
|
||||
print("\n🎉 基础功能测试通过！可以运行主程序了。")
|
||||
if private_ok:
|
||||
print("🔐 API密钥配置正确，可以进行交易操作。")
|
||||
else:
|
||||
print("⚠️ API密钥需要配置或有问题，只能使用公共功能。")
|
||||
else:
|
||||
print("\n❌ 存在配置问题，请检查后重试。")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Loading…
Reference in New Issue