# 天勤量化框架普惠版

**Repository Path**: friendships/tqsdk

## Basic Information

- **Project Name**: 天勤量化框架普惠版
- **Description**: 基于天勤量化API的适配开发，纯python，适合量化新手。
- **Primary Language**: Python
- **License**: MIT
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 2
- **Created**: 2025-12-04
- **Last Updated**: 2025-12-04

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# TqSdk Trader - 稳健优雅的量化交易框架

基于 TqSdk 构建的现代化量化交易框架，采用模块化设计，提供完整的交易生命周期管理。

## 🚀 核心特性

### ✨ 架构优势
- **模块化设计**: 高内聚、低耦合的组件架构
- **异步编程**: 基于 asyncio 的高性能异步处理
- **类型安全**: 完整的类型注解和验证
- **错误处理**: 完善的异常处理和恢复机制

### 🛡️ 风险控制
- **多层风控**: 持仓限制、亏损限制、交易频率控制
- **实时监控**: 连接状态、持仓状态、盈亏监控
- **优雅降级**: 系统异常时的平滑降级处理
- **紧急停止**: 一键停止所有交易活动

### 📊 数据管理
- **智能缓存**: 数据缓存和过期管理
- **实时更新**: 行情和K线数据的实时推送
- **多周期支持**: 支持多种时间周期的数据获取
- **内存优化**: 大数据处理的内存优化

### 🎯 策略支持
- **策略接口**: 统一的策略开发接口
- **双均线策略**: 开箱即用的经典策略实现
- **扩展性**: 易于添加新的交易策略
- **参数配置**: 灵活的策略参数配置

## 📁 项目结构

```
tqsdk_trader/
├── config/                 # 配置管理
│   ├── settings.yaml      # 配置文件
│   └── config_manager.py  # 配置管理器
├── core/                  # 核心模块
│   ├── base_classes.py    # 基础抽象类
│   ├── api_manager.py     # API管理器
│   ├── trading_manager.py # 交易管理器
│   └── ...
├── strategies/            # 策略模块
│   ├── base_strategy.py   # 策略基础类
│   ├── dual_ma_strategy.py # 双均线策略
│   └── ...
├── risk/                  # 风险管理
│   └── risk_manager.py    # 风险管理器
├── data/                  # 数据管理
│   └── data_manager.py    # 数据管理器
├── utils/                 # 工具模块
│   ├── logger.py          # 日志管理
│   ├── exceptions.py      # 异常处理
│   └── ...
├── logs/                  # 日志文件
├── tests/                 # 测试用例
└── main.py               # 主程序入口
```

## ⚡ 快速开始

### 1. 环境准备

```bash
# 安装依赖
pip install tqsdk pyyaml

# 或者使用requirements.txt
pip install -r requirements.txt
```

### 2. 配置设置

编辑 `config/settings.yaml` 文件：

```yaml
# === API配置 ===
api:
  account: "your_account"      # 天勤账户
  password: "your_password"    # 账户密码
  debug: false                 # 调试模式

# === 策略配置 ===
strategy:
  name: "dual_ma_strategy"     # 策略名称
  symbol: "SHFE.rb2401"       # 合约代码
  volume: 3                   # 交易手数
  short_period: 5             # 短期均线周期
  long_period: 20             # 长期均线周期
```

### 3. 运行交易

```bash
# 方式1：直接运行
python main.py

# 方式2：指定配置文件
python main.py config/settings.yaml

# 方式3：在Python代码中调用
from main import run_trader
run_trader("config/settings.yaml")
```

## 📖 使用指南

### 配置管理

框架使用 YAML 配置文件，支持以下配置项：

#### API配置
```yaml
api:
  account: "your_account"      # 天勤账户名
  password: "your_password"    # 账户密码
  url: "https://..."          # 服务器地址（可选）
  debug: false                 # 是否启用调试模式
  web_gui: false              # 是否启用Web界面
```

#### 策略配置
```yaml
strategy:
  name: "strategy_name"       # 策略名称
  symbol: "SHFE.rb2401"       # 合约代码
  volume: 3                   # 每次交易手数
  short_period: 5             # 短期均线周期
  long_period: 20             # 长期均线周期
```

#### 风险管理配置
```yaml
risk:
  max_daily_loss: 5000        # 日最大亏损
  max_single_loss: 1000       # 单次最大亏损
  max_position_size: 10       # 最大持仓手数
  stop_loss_points: 50        # 止损点数
  take_profit_points: 150     # 止盈点数
```

## 🔄 交易完整流程

TqSdk Trader 框架实现了完整的量化交易生命周期管理，从配置加载到平仓执行的全流程自动化。

### 📋 核心交易流程

```mermaid
graph TD
    A[1. 配置加载] --> B[2. API连接]
    B --> C[3. 数据获取]
    C --> D[4. 策略计算]
    D --> E[5. 信号生成]
    E --> F[6. 订单下达]
    F --> G[7. 状态跟踪]
    G --> H[8. 风险控制]
    H --> I[9. 平仓执行]

    I --> J{继续交易?}
    J -->|是| C
    J -->|否| K[结束交易]

    H -.->|触发风控| L[紧急停止]
    L --> K
```

### 🎯 详细流程说明

| 步骤 | 环节名称 | 核心功能 | 涉及模块 |
|-----|---------|---------|---------|
| **1** | **配置加载** | 加载YAML配置文件，初始化系统参数 | `config_manager.py` |
| **2** | **API连接** | 建立与TqSdk服务器的安全连接 | `api_manager.py` |
| **3** | **数据获取** | 获取实时行情、K线数据、Tick数据 | `data_manager.py` |
| **4** | **策略计算** | 执行技术指标计算和策略逻辑 | `dual_ma_strategy.py` |
| **5** | **信号生成** | 根据策略逻辑生成买卖信号 | `base_strategy.py` |
| **6** | **订单下达** | 提交市价单或限价单到交易所 | `trading_manager.py` |
| **7** | **状态跟踪** | 实时监控订单状态和成交情况 | `trading_manager.py` |
| **8** | **风险控制** | 执行止损止盈、持仓限制等风控措施 | `risk_manager.py` |
| **9** | **平仓执行** | 根据信号或风控条件执行平仓操作 | `trading_manager.py` |

### 🔄 循环执行机制

1. **实时数据流**: 步骤3-9形成闭环，不断处理新的市场数据
2. **并发处理**: 多个策略可同时运行，互不干扰
3. **异常处理**: 任何步骤出现异常都会触发错误处理和恢复机制
4. **风控优先**: 步骤8的风险控制贯穿整个交易流程

### 📊 流程监控指标

框架提供完整的流程监控：

- **连接状态**: API连接健康度
- **数据质量**: 数据获取成功率和延迟
- **策略性能**: 胜率、盈亏比、最大回撤
- **订单执行**: 成交率、滑点控制
- **风险指标**: 持仓比例、亏损控制

### 🎯 使用示例

```python
# 完整交易流程示例
from main import TqSdkTrader

# 1. 初始化交易器
trader = TqSdkTrader("config/settings.yaml")

# 2. 启动交易（自动执行完整流程）
await trader.initialize()  # 配置加载 & API连接
await trader.start_trading()  # 数据获取 → 策略计算 → 信号生成 → 订单下达...

# 3. 监控交易状态
status = await trader.get_system_status()
print(f"交易流程状态: {status}")
```

### 🛡️ 异常处理机制

- **连接异常**: 自动重连机制
- **数据异常**: 使用缓存数据或停止策略
- **订单异常**: 撤单重发或报警处理
- **风控触发**: 立即平仓并暂停策略
- **系统异常**: 优雅降级和状态保存

### 📈 性能优化

- **异步处理**: 全流程异步执行，提高并发性能
- **数据缓存**: 智能缓存减少API调用频率
- **批量处理**: 批量数据获取和订单处理
- **内存管理**: 自动清理过期数据和对象

---

## 🎯 策略开发

#### 1. 继承基础策略类

```python
from strategies.base_strategy import BaseStrategy, SignalType

class MyStrategy(BaseStrategy):
    def __init__(self, symbol: str, config: dict):
        super().__init__(symbol, config)

    async def _initialize_strategy(self):
        """初始化策略"""
        # 初始化逻辑
        pass

    async def generate_signal(self) -> SignalType:
        """生成交易信号"""
        # 信号生成逻辑
        return SignalType.BUY  # 或 SignalType.SELL 或 SignalType.HOLD

    async def calculate_position_size(self) -> int:
        """计算仓位大小"""
        return self.config.get('volume', 1)
```

#### 2. 注册和运行策略

```python
from strategies.base_strategy import get_strategy_manager

# 创建策略实例
strategy = MyStrategy("SHFE.rb2401", config)

# 注册到策略管理器
strategy_manager = get_strategy_manager()
await strategy_manager.register_strategy(strategy)

# 启动策略
await strategy_manager.start_strategy("my_strategy", "SHFE.rb2401")
```

### 风险管理

框架内置完善的风险管理机制：

#### 持仓风险检查
```python
from risk.risk_manager import get_risk_manager

risk_manager = get_risk_manager()
risk_info = await risk_manager.check_position_risk(
    symbol="SHFE.rb2401",
    position=5,
    entry_price=3500,
    current_price=3520
)
```

#### 止损止盈设置
```python
# 自动止损点数
stop_loss_points = 50

# 自动止盈点数
take_profit_points = 150

# 跟踪止损
enable_trailing_stop = True
trailing_stop_points = 30
```

### 数据管理

#### 获取实时行情
```python
from data.data_manager import get_market_data_manager

data_manager = get_market_data_manager()
quote = await data_manager.get_quote("SHFE.rb2401")
print(f"最新价: {quote['last_price']}")
```

#### 获取K线数据
```python
kline_data = await data_manager.get_kline_data("SHFE.rb2401", 60, 100)
klines = kline_data['data']
print(f"最新K线收盘价: {klines.iloc[-1]['close']}")
```

### 交易管理

#### 下单交易
```python
from core.trading_manager import get_trading_manager

trading_manager = get_trading_manager()

# 限价单
order_id = await trading_manager.place_limit_order(
    symbol="SHFE.rb2401",
    direction="BUY",
    offset="OPEN",
    volume=5,
    price=3500
)

# 市价单
order_id = await trading_manager.place_market_order(
    symbol="SHFE.rb2401",
    direction="SELL",
    offset="CLOSE",
    volume=3
)
```

#### 订单管理
```python
# 获取活跃订单
active_orders = await trading_manager.get_active_orders()

# 撤单
success = await trading_manager.cancel_order(order_id)

# 批量撤单
cancelled_count = await trading_manager.cancel_all_orders()
```

## 🔧 高级功能

### 自定义日志

框架提供灵活的日志配置：

```python
from utils.logger import get_logger

# 获取指定模块的日志器
logger = get_logger("MyStrategy")

# 记录不同级别的日志
logger.debug("调试信息")
logger.info("普通信息")
logger.warning("警告信息")
logger.error("错误信息")
```

### 性能监控

```python
from main import TqSdkTrader

trader = TqSdkTrader()
status = await trader.get_system_status()

print(f"系统健康状态: {status['system_health']}")
print(f"活跃订单数量: {status['trading_summary']['active_orders']}")
```

### 异步任务管理

框架使用 asyncio 进行异步任务管理：

```python
import asyncio

async def my_async_task():
    # 异步任务逻辑
    await asyncio.sleep(1)
    return "完成"

# 创建任务
task = asyncio.create_task(my_async_task())

# 等待任务完成
result = await task
```

## 🧪 测试

运行测试用例：

```bash
# 运行所有测试
python -m pytest tests/

# 运行特定测试
python -m pytest tests/test_strategy.py

# 带覆盖率报告
python -m pytest --cov=core --cov=strategies tests/
```

## 📊 监控和调试

### 系统状态监控

```python
# 获取系统状态
status = await trader.get_system_status()

# 健康检查
if status['system_health'] != 'healthy':
    print("系统异常，执行降级处理")
```

### 日志分析

```bash
# 查看交易日志
tail -f logs/trader.log

# 搜索特定交易
grep "TRADE" logs/trader.log

# 分析错误日志
grep "ERROR" logs/trader.log
```

## 🚨 故障排除

### 常见问题

#### 1. 连接失败
```python
# 检查网络连接
# 验证账户密码
# 检查服务器地址
```

#### 2. 订单被拒绝
```python
# 检查账户权限
# 验证合约代码
# 检查资金充足性
```

#### 3. 策略不执行
```python
# 检查策略状态
# 验证数据获取
# 查看日志错误信息
```

### 调试模式

启用调试模式获取更多信息：

```yaml
api:
  debug: true

logging:
  level: "DEBUG"
```

## 📈 性能优化

### 配置优化

```yaml
# 数据缓存配置
data:
  cache_enabled: true
  cache_ttl: 300

# 交易频率限制
trading:
  max_orders_per_minute: 10
```

### 内存优化

```python
# 定期清理缓存
await data_manager.invalidate_cache()

# 清理已完成订单
await trading_manager.cleanup_completed_orders()
```

## 🔐 安全考虑

1. **账户安全**: 妥善保管账户密码，避免硬编码
2. **网络安全**: 使用HTTPS连接，确保数据传输安全
3. **权限控制**: 合理配置账户权限，最小化风险
4. **日志安全**: 注意日志中敏感信息的安全处理

## 🤝 贡献指南

### 开发流程

1. Fork 项目
2. 创建特性分支
3. 编写测试用例
4. 实现功能代码
5. 运行测试验证
6. 提交 Pull Request

### 代码规范

- 使用类型注解
- 编写详细的文档字符串
- 添加必要的异常处理
- 遵循 PEP 8 编码规范

## 📄 许可证

本项目采用 MIT 许可证。

## 📞 联系方式

如有问题或建议，请：

1. 查看项目文档
2. 提交 Issue
3. 联系项目维护者

## 📈 性能特点

### 交易耗时分析
```
总耗时: 50-200毫秒 (典型情况)
├── 数据接收: 10-50ms   (网络传输 + 协议解析)
├── 策略计算: 5-20ms    (K线处理 + 指标计算)
├── 信号生成: 1-5ms     (逻辑判断)
└── 订单下达: 30-150ms  (网络传输 + 撮合)
```

### 性能优势
- ✅ **异步架构**: 全流程异步处理，无阻塞等待
- ✅ **智能缓存**: 减少重复数据请求和计算
- ✅ **向量化计算**: pandas/numpy高效数据处理
- ✅ **连接复用**: 减少网络连接建立耗时

### 适用场景
- ✅ **中低频策略**: 完全满足 (日间交易、波段交易)
- ✅ **高频策略**: 基本满足 (需优化网络延迟)
- ✅ **多策略组合**: 完全支持 (并发处理)

---

## 🎉 项目总结

**TqSdk Trader** 是一个**生产级别**的量化交易框架，具有完整的交易生命周期管理能力。

### 🎯 核心特性
- ✅ **完整的交易生命周期管理**
- ✅ **模块化架构设计**
- ✅ **异步高性能处理**
- ✅ **企业级风险控制**
- ✅ **可扩展策略框架**
- ✅ **完善的监控日志**

### 🚀 立即可用
框架已具备**生产环境部署条件**，可以：
- 直接用于实盘交易
- 扩展添加新策略
- 集成监控告警系统
- 部署到服务器环境

---

**享受量化交易的乐趣！** 🎯📈