量化交易系统 - 持仓分析模块

项目概述

主要功能

1. 回测框架

2. 持仓分析模块

2. 技术分析模块

3. XtQuant集成

Python环境

conda create -n py311 python=3.11.9
conda activate py311

安装依赖

pip install -r requirements.txt

环境配置

1. .env 环境变量

# 复制环境变量示例文件
cp .env.example .env
# 编辑 .env 文件,配置数据服务、数据库等运行所需的敏感信息
# 例如:TUSHARE_TOKEN、MYSQL_HOST、MYSQL_USER、MYSQL_PASSWORD 等

2. config/Config.yaml 交易环境

快速开始

1. 启动Web服务器

python main.py

2. 运行演示脚本

# 持仓分析演示
python demo_position_analysis.py
# XtQuant技术分析演示
python demo_xtquant_analysis.py
# Backtrader 回测演示
python demo_backtrader_example.py
# Backtrader 框架测试
python test_backtrader_framework.py
# 比亚迪策略回测演示
python simple_byd_backtest.py
# 比亚迪策略演示
python demo_byd_strategy.py
# 比亚迪策略简单测试
python test_byd_simple.py

3. 访问API接口

API 文档 (Swagger)

API端点列表

持仓分析API

技术分析API

使用示例

1. 持仓分析

from modules.tornadoapp.position.position_analyzer import PositionAnalyzer
# 创建分析器
analyzer = PositionAnalyzer(tushare_token)
# 分析持仓
positions_data = [
 {
 "symbol": "000001.SZ",
 "volume": 1000,
 "avg_price": 15.50,
 "current_price": 16.20
 }
]
analysis = analyzer.analyze_positions(positions_data, cash=50000)
print(f"总收益率: {analysis.summary.total_unrealized_pnl_pct:.2f}%")

2. 回测分析

from modules.backtrader_engine import BacktraderEngine, TushareDataFeed, MAStrategy
# 创建回测引擎
engine = BacktraderEngine(initial_cash=1000000, commission=0.001)
# 添加数据源和策略
data_feed = TushareDataFeed(symbol='000001.SZ', start_date='20230101', 
 end_date='20231231', tushare_token='your_token')
engine.add_data(data_feed)
engine.add_strategy(MAStrategy)
# 运行回测
result = engine.run_backtest()
print(f"总收益率: {result.total_return:.2%}")
print(f"夏普比率: {result.sharpe_ratio:.2f}")

3. 比亚迪策略回测

from strategies.byd_strategy import BYDStrategy, BYDEnhancedStrategy, BYDConservativeStrategy
# 运行比亚迪策略回测
python simple_byd_backtest.py
# 查看详细说明
# 参考 BYD_STRATEGY_SUMMARY.md 文件

3. 技术分析

from modules.tornadoapp.position.xtquant_position_manager import XtQuantPositionManager
# 创建XtQuant管理器
manager = XtQuantPositionManager(tushare_token)
# 分析持仓
results = await manager.analyze_all_positions("demo_account")
# 获取交易建议
recommendations = manager.generate_trading_recommendations(results)
for rec in recommendations:
 print(f"{rec['symbol']}: {rec['action']} (置信度: {rec['confidence']:.2f})")

3. 可视化报告

from modules.tornadoapp.position.position_visualizer import PositionVisualizer
# 创建可视化工具
visualizer = PositionVisualizer()
# 生成图表
charts = visualizer.generate_comprehensive_report(analysis)

项目结构

modules/tornadoapp/position/
├── __init__.py # 模块初始化
├── position_analyzer.py # 持仓分析器
├── position_visualizer.py # 可视化工具
├── xtquant_position_manager.py # XtQuant集成
└── position_api.py # API路由
modules/tornadoapp/handler/
├── position_handler.py # 持仓分析API处理器
└── technical_analysis_handler.py # 技术分析API处理器
modules/tornadoapp/model/
└── position_model.py # 持仓数据模型

技术指标说明

支持的指标

交易信号生成

实盘记录

风险提示

开发说明

添加新的技术指标

def calculate_custom_indicator(self, df: pd.DataFrame) -> float:
 """计算自定义指标"""
 close_prices = df['close'].values.astype(np.float64)
 # 实现指标计算逻辑
 return indicator_value

扩展交易信号逻辑

def generate_custom_signals(self, indicators: Dict, current_price: float) -> Dict:
 """生成自定义交易信号"""
 # 实现信号生成逻辑
 return signals

市场数据存储

数据库设计

数据库特性

核心表结构

1. 股票基础信息

2. 市场数据表

3. 技术指标

4. 数据管理

快速开始

1. 初始化数据库

# 一键初始化数据库
python database/init_market_data.py --host localhost --user root --password your_password
# 测试连接
python database/init_market_data.py --test-only --host localhost --user root --password your_password

2. 基本使用

from database.market_data_manager import MarketDataManager
# 创建数据管理器
manager = MarketDataManager(
 host='localhost',
 port=3306,
 user='market_data_app',
 password='app_password',
 database='market_data'
)
# 测试连接
if manager.test_connection():
 print("✅ 数据库连接成功!")

3. 存储市场数据

import pandas as pd
# 存储股票基础信息
stocks_data = pd.DataFrame({
 'ts_code': ['000001.SZ', '000002.SZ', '600519.SH'],
 'symbol': ['000001', '000002', '600519'],
 'name': ['平安银行', '万科A', '贵州茅台'],
 'industry': ['银行', '房地产', '食品饮料'],
 'market': ['主板', '主板', '主板']
})
manager.insert_stock_basic(stocks_data)
# 存储日线数据
daily_data = pd.DataFrame({
 'symbol': ['000001.SZ'],
 'trade_date': ['2025-01-15'],
 'open': [15.50], 'high': [15.80], 'low': [15.40], 'close': [15.70],
 'volume': [1000000], 'amount': [15700000], 'pct_chg': [1.29]
})
manager.insert_daily_data(daily_data)
# 存储tick数据
tick_data = pd.DataFrame({
 'symbol': ['000001.SZ'] * 100,
 'tick_time': pd.date_range('2025-01-15 09:30:00', periods=100, freq='3S'),
 'last_price': [15.50 + i * 0.01 for i in range(100)],
 'volume': [1000] * 100,
 'amount': [15500 + i * 10 for i in range(100)]
})
manager.insert_tick_data(tick_data)

4. 查询市场数据

# 获取股票列表
stocks = manager.get_stock_list()
print(f"股票总数: {len(stocks)}")
# 获取日线数据
daily_data = manager.get_daily_data('000001.SZ', '2025-01-01', '2025-01-15')
print(f"日线数据条数: {len(daily_data)}")
# 获取tick数据
tick_data = manager.get_tick_data('000001.SZ', '2025-01-15 09:30:00', '2025-01-15 10:00:00')
print(f"tick数据条数: {len(tick_data)}")
# 获取最新价格
latest_price = manager.get_latest_price('000001.SZ')
print(f"平安银行最新价格: {latest_price}")

5. 为AI模型准备数据

# 获取训练数据
def get_training_data(symbol, start_date, end_date):
 """获取用于AI模型训练的数据"""
 # 获取日线数据
 daily_data = manager.get_daily_data(symbol, start_date, end_date)
 
 # 获取技术指标
 indicators = manager.get_technical_indicators(symbol, start_date, end_date)
 
 # 合并数据
 training_data = daily_data.merge(indicators, on=['symbol', 'trade_date'], how='left')
 
 # 计算特征
 training_data['price_change'] = training_data['close'].pct_change()
 training_data['volume_change'] = training_data['volume'].pct_change()
 training_data['ma5_ma20_diff'] = training_data['ma5'] - training_data['ma20']
 
 return training_data
# 获取训练数据
training_data = get_training_data('000001.SZ', '2024-01-01', '2024-12-31')
print(f"训练数据形状: {training_data.shape}")

数据导出

# 导出数据到CSV
csv_file = manager.export_data_to_csv(
 symbol='000001.SZ',
 data_type='daily',
 start_date='2025-01-01',
 end_date='2025-01-15',
 output_dir='data/export'
)
print(f"数据导出到: {csv_file}")

详细使用指南

许可证

联系方式

添加微信好友

扫一扫上面的二维码图案,加我为朋友。

加入量化投资交流群

群聊:量化投资交流群

⚠️ 该二维码7天内(3月16日前)有效,重新进入将更新