From d1079d338d3f512a5bfa825b3abbfb64d5b92ba9 Mon Sep 17 00:00:00 2001 From: blade <8019068@qq.com> Date: Mon, 21 Jul 2025 13:05:59 +0800 Subject: [PATCH] first --- .cursorrules | 42 ++++ QUICKSTART.md | 94 ++++++++ README.md | 164 ++++++++++++++ __pycache__/config.cpython-313.pyc | Bin 0 -> 730 bytes config.py | 37 ++++ demo.py | 306 ++++++++++++++++++++++++++ play.py | 334 +++++++++++++++++++++++++++++ requirements.txt | 4 + test_connection.py | 129 +++++++++++ 9 files changed, 1110 insertions(+) create mode 100644 .cursorrules create mode 100644 QUICKSTART.md create mode 100644 README.md create mode 100644 __pycache__/config.cpython-313.pyc create mode 100644 config.py create mode 100644 demo.py create mode 100644 play.py create mode 100644 requirements.txt create mode 100644 test_connection.py diff --git a/.cursorrules b/.cursorrules new file mode 100644 index 0000000..74563e0 --- /dev/null +++ b/.cursorrules @@ -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开发最佳实践。 \ No newline at end of file diff --git a/QUICKSTART.md b/QUICKSTART.md new file mode 100644 index 0000000..4e2455b --- /dev/null +++ b/QUICKSTART.md @@ -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. 交易所服务是否可用 \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..9f0f3d4 --- /dev/null +++ b/README.md @@ -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 \ No newline at end of file diff --git a/__pycache__/config.cpython-313.pyc b/__pycache__/config.cpython-313.pyc new file mode 100644 index 0000000000000000000000000000000000000000..e6a23047265e5322344c51042765297fff491f32 GIT binary patch literal 730 zcmZ9JL2nX49K{DH1s4_yU8tbxrD+-t5-(m%Bf*wd(i91U2b0Fh!cLcwWoI)pkZz3$ zCVm7x=r`cv-MgQ_(`IhQ6E{e&9-O5mCjOIs`+F}l`M>Pm=(+~foc(w|IEw@DJ2dOb z+`-MEg3kZ~24I9mV3b8+3~sMRDgU=HTtb~VN}Y@AT*Y&5`9ek|WU%!MiTf_HSMVFi0AN8msb&ql)2z3!B*H|XF z=~_K{CAavjqNif_iEN=g8{|nHA;`BahKbqDu6|Jp2(d`eZb>6jWj|#|1M5 zwtYlgf!V}#WM=3P7yt9ATknt&5v@vGChF}bI%~X^8f+YO8r@rN*CA}b=+t3DX1bmF zakF)bj$XD-nx`_=ZJsyo%naP9#Xue$KfUmLF;E{(wR?;Ef?s%!>D!EsUF)Jx-AiIu zJzpjdN7NcS_!-+*_af@yZ= 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() \ No newline at end of file diff --git a/play.py b/play.py new file mode 100644 index 0000000..515f9e1 --- /dev/null +++ b/play.py @@ -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() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..142ea21 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +okx>=2.1.2 +pandas>=2.0.0 +numpy>=1.20.0 +requests>=2.25.0 \ No newline at end of file diff --git a/test_connection.py b/test_connection.py new file mode 100644 index 0000000..cf7f11a --- /dev/null +++ b/test_connection.py @@ -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() \ No newline at end of file