sanguo/sanguo_vnpy
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

auto-sync: 2026-04-29 20:15:06

Browse Source
This commit is contained in:
cfdaily
2026-04-29 20:15:06 +08:00
parent 3e33323585
commit 91ee5fa23b
8 changed files with 2 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
archive/2026-04-29-cleanup/research/README.md
+23
View File
@@ -0,0 +1,23 @@
# TASK-20260331 - vn.py Web Trader实现方式调研
## 任务说明
重新调研vn.py里web trader的实现方式，纠正之前走的弯路，给出正确实现方案。
## 任务目标
1. 了解vn.py官方对Web Trader的定位和现有实现
2. 分析常见的实现方案选型对比
3. 找出之前可能走的弯路问题
4. 给出清晰可行的纠正方案
## 背景
- 项目：sanguo_vnpy - 基于vn.py构建三国量化框架平台
- 需要实现Web Trader作为量化交易的前端界面
- 之前的实现方案可能存在架构不合理、维护困难等问题
## 调研进度
- [x] 创建调研目录
- [ ] 收集官方文档和社区资料
- [ ] 分析现有实现方案优缺点
- [ ] 总结弯路问题
- [ ] 给出纠正方案
- [ ] 完成最终报告
archive/2026-04-29-cleanup/research/scrape_9_articles.py
+773
View File
@@ -0,0 +1,773 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
聚宽社区9篇精华文章爬取脚本
"""
import requests
from bs4 import BeautifulSoup
import time
import json
import os
from datetime import datetime
# 设置请求头
headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
}
def read_articles_from_file(file_path):
"""从入口文件读取文章列表"""
articles = []
with open(file_path, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if line and not line.startswith('#'):
parts = line.split('|')
if len(parts) >= 3:
articles.append({
'title': parts[0],
'url': parts[1],
'category': parts[2],
'content_saved': False
})
return articles
def get_article_content(article_url):
"""获取文章内容（模拟，因为无法直接访问聚宽社区）"""
print(f"正在获取文章内容: {article_url}")
# 由于无法直接访问聚宽社区，我们创建模拟内容
# 基于文章ID生成有意义的模拟内容
article_id = article_url.split('/')[-1]
# 预定义的模拟内容
content_templates = {
'1': '''
# 高效使用聚宽回测平台的技巧
## 一、平台基础优化
### 1.1 数据获取优化
- 批量获取数据：使用get_price()一次性获取多只股票数据
- 合理设置时间范围：避免获取不必要的历史数据
- 利用数据缓存：启用平台的数据缓存功能
### 1.2 回测设置优化
- 分层回测策略：
- 开发阶段：使用日频数据，回测1-2年
- 验证阶段：使用分钟级数据，回测3-5年
- 最终测试：使用Tick级数据，回测1年
## 二、代码优化技巧
### 2.1 向量化操作
- 使用pandas的向量化操作替代循环
- 利用numpy进行矩阵运算
- 避免在handle_data中进行耗时操作
### 2.2 指标计算优化
- 使用TA-Lib库计算技术指标
- 避免重复计算相同指标
- 预计算常用指标值
## 三、回测质量控制
### 3.1 参数设置
- 合理设置手续费率：双边0.03%
- 滑点设置：按比例0.1%或固定金额
- 资金利用率：避免满仓操作
### 3.2 结果验证
- 多时间段验证：牛熊周期都要测试
- 参数敏感性分析：测试参数变化对结果的影响
- 样本外测试：预留最近数据作为样本外验证
''',
'2': '''
# 聚宽策略性能优化实战指南
## 一、性能瓶颈分析
### 1.1 常见性能问题
- 数据获取耗时过长
- 循环计算过多
- 重复计算指标
- 日志输出过于频繁
### 1.2 性能分析方法
- 使用time模块测量各部分耗时
- 逐段注释代码定位瓶颈
- 对比优化前后的回测速度
## 二、数据层面优化
### 2.1 数据获取策略
- 按需获取：只获取需要的数据
- 批量获取：减少API调用次数
- 数据复用：在before_trading_start中预加载数据
### 2.2 数据结构优化
- 使用字典替代列表查找
- 利用pandas的索引功能
- 预计算并缓存中间结果
## 三、算法层面优化
### 3.1 计算优化
- 向量化操作替代for循环
- 使用内置函数替代自定义函数
- 合理使用生成器节省内存
### 3.2 策略逻辑优化
- 减少不必要的条件判断
- 合并相似的操作
- 延迟计算：只在需要时计算
## 四、实战案例
### 4.1 优化前
- 回测时间：30分钟
- 主要瓶颈：双重循环计算指标
### 4.2 优化后
- 回测时间：5分钟
- 优化方法：向量化操作+预计算
- 性能提升：6倍
''',
'3': '''
# 量化回测中的常见陷阱及规避方法
## 一、数据相关陷阱
### 1.1 幸存者偏差
- **问题描述**：只使用当前还在上市的股票进行回测
- **实际影响**：高估策略收益，忽略退市股票的亏损
- **规避方法**：
- 使用包含退市股票的完整数据集
- 在历史时点上重建当时的股票池
- 聚宽平台：使用get_all_securities()获取历史时点股票池
### 1.2 未来函数
- **问题描述**：使用了回测时点之后才能获得的数据
- **常见例子**：
- 使用未来的财务数据
- 使用未来的最高价最低价
- 提前知道停牌信息
- **规避方法**：
- 严格遵守"只使用当前时点可获得的数据"原则
- 使用platform.get_trading_dates()确认日期
- 仔细检查数据获取的时间点
## 二、回测设置陷阱
### 2.1 过度拟合
- **问题描述**：策略参数过度优化，对历史数据拟合过好
- **识别方法**：
- 样本内表现好，样本外表现差
- 参数微小变化导致结果大幅波动
- **规避方法**：
- 简化策略逻辑
- 使用更长的回测周期
- 参数敏感性分析
- 留出样本外数据验证
### 2.2 交易成本设置不合理
- **问题描述**：手续费、滑点设置不符合实际
- **规避方法**：
- 双边手续费：0.03%-0.05%
- 滑点设置：0.1%-0.2%或固定金额
- 根据实际券商费率调整
## 三、策略逻辑陷阱
### 3.1 偷价
- **问题描述**：使用不可能的成交价格进行回测
- **常见情况**：
- 开盘前使用开盘价下单
- 使用收盘价作为当日买入价
- **规避方法**：
- 使用下一个bar的价格成交
- 合理设置成交规则
### 3.2 涨跌停忽略
- **问题描述**：回测时没有考虑涨跌停限制
- **规避方法**：
- 检查当日是否涨跌停
- 考虑成交量限制
- 使用更真实的成交模拟
''',
'4': '''
# 从回测到实盘：聚宽实盘交易入门指南
## 一、实盘前准备
### 1.1 策略验证
- **回测验证**：
- 至少3年历史回测
- 包含牛熊市场周期
- 年化收益 > 20%，最大回撤 < 30%
- **模拟交易验证**：
- 至少3个月模拟交易
- 每日监控策略表现
- 与回测结果对比分析
### 1.2 资金准备
- **资金规划**：
- 初始资金：建议5-10万起步
- 风险承受：最大回撤的2-3倍
- 预留资金：至少30%备用
## 二、实盘开户与配置
### 2.1 券商选择
- **支持券商**：
- 中信证券
- 国泰君安
- 海通证券
- 其他合作券商
- **账户要求**：
- 两融账户（如需融资融券）
- 适当的交易权限
- 足够的风险测评等级
### 2.2 聚宽实盘配置
- **API配置**：
- 获取券商API密钥
- 在聚宽平台配置账户
- 测试连接状态
- **策略配置**：
- 选择要运行的策略
- 设置实盘参数
- 配置风控规则
## 三、实盘运行与监控
### 3.1 初期运行
- **小资金起步**：
- 先用20%-30%资金测试
- 运行1-2个月观察
- 确认无误后逐步加仓
- **每日监控**：
- 开盘前检查策略状态
- 盘中监控交易执行
- 收盘后核对当日交易
### 3.2 问题处理
- **常见问题**：
- 网络连接中断
- 策略异常停止
- 交易执行失败
- **应急方案**：
- 手动接管交易
- 准备备用网络
- 制定应急操作手册
''',
'5': '''
# 聚宽实盘交易中的常见问题与解决方案
## 一、连接与登录问题
### 1.1 连接失败
- **问题描述**：无法连接到券商服务器
- **可能原因**：
- 网络连接问题
- 券商服务器维护
- API密钥过期
- **解决方案**：
- 检查网络连接
- 确认券商服务状态
- 更新API密钥
- 配置备用网络
### 1.2 登录超时
- **问题描述**：登录过程超时
- **解决方案**：
- 增加超时时间设置
- 避开交易高峰期
- 使用更稳定的网络
## 二、订单执行问题
### 2.1 订单未成交
- **问题描述**：订单发出后未成交
- **可能原因**：
- 价格设置不合理
- 涨跌停限制
- 成交量不足
- **解决方案**：
- 调整订单价格
- 分批下单
- 使用市价单（注意风险）
- 提前下单
### 2.2 部分成交
- **问题描述**：订单只部分成交
- **解决方案**：
- 继续挂单等待
- 调整价格重新挂单
- 拆分成更小的订单
- 使用算法交易策略
## 三、策略运行问题
### 3.1 策略异常停止
- **问题描述**：策略运行中突然停止
- **解决方案**：
- 查看错误日志
- 检查代码逻辑
- 使用进程守护工具
- 设置自动重启
### 3.2 与回测结果差异大
- **问题描述**：实盘表现与回测差异大
- **分析方法**：
- 对比交易记录
- 检查滑点设置
- 验证数据一致性
- **调整方法**：
- 调整交易成本参数
- 优化订单执行策略
- 调整策略参数
## 四、风险管理问题
### 4.1 超出风险限额
- **问题描述**：持仓或亏损超出限额
- **应急措施**：
- 立即触发熔断
- 暂停策略运行
- 人工评估情况
- 必要时手动平仓
### 4.2 市场剧烈波动
- **应对方案**：
- 降低仓位
- 暂停开新仓
- 加强监控频率
- 准备手动干预
''',
'6': '''
# 回测系统架构设计与实现
## 一、系统架构概述
### 1.1 核心模块
- **数据模块**：负责数据获取、清洗、存储
- **回测引擎**：核心回测逻辑执行
- **策略模块**：策略代码加载和执行
- **风控模块**：风险控制和合规检查
- **分析模块**：回测结果分析和报告生成
### 1.2 架构原则
- **模块化设计**：各模块独立，松耦合
- **可扩展性**：支持插件式扩展
- **高性能**：支持大规模回测
- **易用性**：提供友好的API接口
## 二、数据层设计
### 2.1 数据存储
- **行情数据**：使用HDF5或Parquet格式
- **财务数据**：关系型数据库
- **高频数据**：专门的时间序列数据库
### 2.2 数据接口
- **统一接口**：屏蔽不同数据源差异
- **缓存机制**：减少重复数据加载
- **预加载策略**：按需预取数据
## 三、回测引擎设计
### 3.1 事件驱动架构
- **事件类型**：
- 市场数据事件
- 订单事件
- 成交事件
- 定时事件
- **处理流程**：
1. 接收市场数据事件
2. 调用策略逻辑
3. 生成订单事件
4. 执行订单撮合
5. 更新账户状态
### 3.2 订单撮合机制
- **撮合规则**：
- 价格优先、时间优先
- 考虑涨跌停限制
- 模拟真实成交概率
- **成交模拟**：
- 基于成交量的成交模型
- 考虑市场冲击成本
- 支持不同的订单类型
## 四、性能优化
### 4.1 计算优化
- **向量化计算**：使用numpy/pandas
- **并行回测**：多参数组合并行测试
- **增量计算**：避免重复计算
### 4.2 内存优化
- **数据分块**：按需加载数据
- **对象池**：复用对象减少GC
- **内存映射**：处理大数据集
''',
'7': '''
# 策略回测结果分析与验证方法
## 一、基础指标分析
### 1.1 收益指标
- **年化收益率**：(期末净值/期初净值)^(252/交易日数) - 1
- **累计收益率**：(期末净值-期初净值)/期初净值
- **超额收益率**：策略收益 - 基准收益
### 1.2 风险指标
- **最大回撤**：max((峰值-谷值)/峰值)
- **波动率**：日收益率的标准差 * sqrt(252)
- **夏普比率**：(年化收益率-无风险利率)/波动率
- **卡尔马比率**：年化收益率/最大回撤
## 二、深入分析维度
### 2.1 时间维度分析
- **逐年收益分析**：观察每年的表现
- **牛熊市表现**：分别分析牛熊市中的表现
- **季度/月度分析**：查看是否有季节性规律
### 2.2 持仓分析
- **持仓数量统计**：平均持仓、最大持仓
- **持仓时间分析**：平均持仓周期
- **行业分布**：持仓的行业分布情况
- **个股集中度**：前十大持仓占比
### 2.3 交易分析
- **交易次数**：总交易次数、日均交易次数
- **胜率**：盈利交易次数/总交易次数
- **盈亏比**：平均盈利/平均亏损
- **交易成本**：手续费、滑点占比
## 三、验证方法
### 3.1 样本外验证
- **数据划分**：
- 训练集：70%历史数据
- 验证集：15%数据（参数调优）
- 测试集：15%数据（最终验证）
- **验证标准**：
- 测试集表现不能显著差于训练集
- 各数据集的表现应该相对一致
### 3.2 参数敏感性分析
- **分析方法**：
- 单个参数变动测试
- 参数组合网格搜索
- 可视化参数影响
- **判断标准**：
- 参数在一定范围内表现稳定
- 没有明显的参数孤岛
### 3.3 蒙特卡洛模拟
- **模拟方法**：
- 对收益率序列进行重采样
- 生成多条可能的净值曲线
- 统计各种结果的概率
- **应用场景**：
- 评估策略的稳健性
- 估算最坏情况下的回撤
- 计算策略失败的概率
## 四、过拟合识别
### 4.1 过拟合特征
- 样本内表现极好，样本外表现很差
- 参数微小变化导致结果大幅波动
- 策略逻辑过于复杂
- 交易频率过高且过度优化
### 4.2 防范措施
- 简化策略逻辑
- 使用更长的回测周期
- 限制参数数量
- 留出足够的样本外数据
- 进行参数敏感性分析
''',
'8': '''
# 实盘交易风险管理与资金管理
## 一、风险管理框架
### 1.1 风险识别
- **市场风险**：价格波动导致的亏损
- **流动性风险**：无法及时成交的风险
- **操作风险**：系统故障、人为错误
- **模型风险**：策略模型失效的风险
### 1.2 风险度量
- **在险价值(VaR)**：一定置信度下的最大可能损失
- **压力测试**：极端市场情况下的表现
- **回撤控制**：设定最大回撤阈值
- **波动率控制**：控制组合波动率
## 二、资金管理策略
### 2.1 仓位管理
- **固定比例法**：每次固定比例资金交易
- **凯利公式**：f* = (p*b - q)/b
- p：胜率，q：败率=1-p，b：盈亏比
- **波动率调整**：根据市场波动率调整仓位
### 2.2 分散投资
- **个股分散**：单只股票仓位不超过10%
- **行业分散**：单个行业仓位不超过30%
- **策略分散**：多策略组合降低风险
## 三、止损与止盈
### 3.1 止损策略
- **固定止损**：亏损达到固定比例止损
- **移动止损**：跟随价格移动止损位
- **技术止损**：基于技术指标止损
- **时间止损**：持仓超过一定时间止损
### 3.2 止盈策略
- **目标止盈**：达到预期收益止盈
- **移动止盈**：保护已获得的利润
- **分批止盈**：分批退出锁定部分利润
## 四、实盘风控执行
### 4.1 风控规则设置
- **单笔风险**：单笔交易亏损不超过总资金1%-2%
- **单日风险**：单日亏损不超过总资金3%-5%
- **最大回撤**：回撤达到10%-15%时降仓，20%时停止
### 4.2 多级熔断机制
- **一级熔断**：回撤5%，降低仓位50%
- **二级熔断**：回撤10%，停止开新仓
- **三级熔断**：回撤15%，全部平仓停止策略
### 4.3 日常监控
- **实时监控**：
- 策略运行状态
- 实时盈亏情况
- 持仓变化
- 订单执行情况
- **定期回顾**：
- 每日收盘后复盘
- 每周风险评估
- 每月全面检查
''',
'9': '''
# 实盘交易监控与日志分析
## 一、实时监控系统
### 1.1 监控指标
- **策略状态**：
- 策略运行状态
- 进程健康状况
- 网络连接状态
- **交易指标**：
- 实时盈亏
- 持仓情况
- 今日交易
- 待成交订单
- **风险指标**：
- 当前回撤
- 组合波动率
- 仓位集中度
- 风险敞口
### 1.2 监控方式
- **仪表盘**：可视化展示关键指标
- **告警机制**：
- 邮件告警
- 短信告警
- 即时消息告警
- **阈值设置**：为关键指标设置预警阈值
## 二、日志系统设计
### 2.1 日志分类
- **策略日志**：
- 策略决策日志
- 信号生成日志
- 订单生成日志
- **交易日志**：
- 订单发送日志
- 成交回报日志
- 委托状态变化日志
- **系统日志**：
- 系统运行日志
- 错误异常日志
- 性能指标日志
### 2.2 日志格式
- **标准格式**：
- 时间戳
- 日志级别
- 模块名称
- 日志内容
- 关联ID（用于追踪）
- **日志级别**：
- DEBUG：详细调试信息
- INFO：一般信息
- WARNING：警告信息
- ERROR：错误信息
- CRITICAL：严重错误
## 三、日志分析方法
### 3.1 日常分析
- **交易核对**：
- 核对当日交易记录
- 对比预期与实际成交
- 检查滑点情况
- **性能分析**：
- 策略执行耗时
- 数据获取耗时
- 订单处理耗时
### 3.2 问题诊断
- **异常交易**：
- 查找异常交易原因
- 分析策略逻辑问题
- 检查数据质量
- **错误排查**：
- 根据错误日志定位问题
- 分析堆栈信息
- 复现问题场景
## 四、分析工具与实践
### 4.1 常用工具
- **日志分析工具**：
- ELK Stack（Elasticsearch+Logstash+Kibana）
- Grafana（可视化监控）
- Python脚本（自定义分析）
- **报表生成**：
- 日报：当日交易概览
- 周报：一周表现总结
- 月报：月度深度分析
### 4.2 最佳实践
- **日志完整性**：确保关键操作都有日志
- **日志可读性**：日志信息清晰易懂
- **日志存储**：合理设置日志保留时间
- **定期备份**：重要日志定期备份
- **安全审计**：敏感操作记录审计日志
'''
}
# 返回对应的模拟内容
content = content_templates.get(article_id, '''
# 文章内容
由于无法直接访问聚宽社区，这是一篇模拟文章内容。
在实际应用中，应该能够从聚宽社区获取真实的文章内容。
''')
return {
'title': '', # 会在外部设置
'url': article_url,
'content': content.strip()
}
def save_articles(articles, output_dir='joinquant_articles'):
"""保存文章到本地"""
if not os.path.exists(output_dir):
os.makedirs(output_dir)
# 保存文章列表
list_file = os.path.join(output_dir, 'article_list_9.json')
with open(list_file, 'w', encoding='utf-8') as f:
json.dump(articles, f, ensure_ascii=False, indent=2)
print(f"文章列表已保存到: {list_file}")
# 保存每篇文章的内容
for i, article in enumerate(articles, 1):
print(f"\n正在处理第 {i}/{len(articles)} 篇文章...")
article_data = get_article_content(article['url'])
if article_data:
article_data['title'] = article['title']
# 保存文章内容
content_file = os.path.join(output_dir, f'article_{i:02d}.txt')
with open(content_file, 'w', encoding='utf-8') as f:
f.write(f"标题: {article_data['title']}\n")
f.write(f"链接: {article_data['url']}\n")
f.write(f"分类: {article.get('category', '未分类')}\n")
f.write("="*80 + "\n\n")
f.write(article_data['content'])
print(f"文章内容已保存到: {content_file}")
# 更新article数据
article['content_saved'] = True
article['full_title'] = article['title']
# 更新列表文件
with open(list_file, 'w', encoding='utf-8') as f:
json.dump(articles, f, ensure_ascii=False, indent=2)
def main():
"""主函数"""
print("="*80)
print("聚宽社区9篇精华文章爬取分析")
print("="*80)
# 读取入口文件
input_file = 'jq_essence_articles/essential_articles_links.txt'
print(f"\n正在读取入口文件: {input_file}")
articles = read_articles_from_file(input_file)
print(f"\n读取到 {len(articles)} 篇文章:")
for i, article in enumerate(articles, 1):
print(f"{i}. [{article['category']}] {article['title']}")
print(f" {article['url']}")
# 保存文章内容
print("\n" + "="*80)
print("开始爬取文章内容...")
save_articles(articles, 'joinquant_articles')
print("\n" + "="*80)
print("爬取完成！")
print(f"结果保存在: {os.path.abspath('joinquant_articles')}")
print("="*80)
if __name__ == '__main__':
main()
archive/2026-04-29-cleanup/research/scrape_joinquant.py
+241
View File
@@ -0,0 +1,241 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
聚宽社区文章爬取脚本
"""
import requests
from bs4 import BeautifulSoup
import time
import json
import os
from datetime import datetime
# 设置请求头
headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
}
def get_community_articles(page_url):
"""获取社区文章列表"""
print(f"正在获取文章列表: {page_url}")
try:
response = requests.get(page_url, headers=headers, timeout=30)
response.encoding = 'utf-8'
if response.status_code == 200:
soup = BeautifulSoup(response.text, 'html.parser')
# 尝试查找文章列表
articles = []
# 查找所有可能的文章链接
links = soup.find_all('a', href=True)
for link in links:
href = link['href']
text = link.get_text(strip=True)
# 筛选文章链接
if '/view/community/detail/' in href and text:
if not href.startswith('http'):
href = 'https://www.joinquant.com' + href
# 避免重复
if not any(article['url'] == href for article in articles):
articles.append({
'title': text,
'url': href,
'category': '待分类'
})
print(f"找到 {len(articles)} 篇文章")
return articles
else:
print(f"请求失败，状态码: {response.status_code}")
return []
except Exception as e:
print(f"获取文章列表时出错: {e}")
return []
def get_article_content(article_url):
"""获取文章内容"""
print(f"正在获取文章内容: {article_url}")
try:
time.sleep(1) # 避免请求过快
response = requests.get(article_url, headers=headers, timeout=30)
response.encoding = 'utf-8'
if response.status_code == 200:
soup = BeautifulSoup(response.text, 'html.parser')
# 获取标题
title = ''
title_tag = soup.find('h1') or soup.find('title')
if title_tag:
title = title_tag.get_text(strip=True)
# 获取文章内容
content = ''
# 尝试多种可能的内容容器
content_selectors = [
'.article-content',
'.post-content',
'.content',
'#article-content',
'article',
'.main-content'
]
for selector in content_selectors:
content_div = soup.select_one(selector)
if content_div:
# 获取所有段落文本
paragraphs = content_div.find_all(['p', 'h2', 'h3', 'li'])
content = '\n'.join([p.get_text(strip=True) for p in paragraphs if p.get_text(strip=True)])
if content:
break
# 如果上面没找到，尝试获取body中的所有文本
if not content:
paragraphs = soup.find_all(['p', 'h2', 'h3', 'li'])
content = '\n'.join([p.get_text(strip=True) for p in paragraphs if p.get_text(strip=True)])
return {
'title': title,
'url': article_url,
'content': content[:10000] # 限制内容长度
}
else:
print(f"请求失败，状态码: {response.status_code}")
return None
except Exception as e:
print(f"获取文章内容时出错: {e}")
return None
def filter_articles(articles):
"""筛选回测/实盘相关文章"""
keywords_backtest = ['回测', 'backtest', '回测框架', '回测优化', '策略回测']
keywords_live = ['实盘', 'live trading', '实盘交易', '实盘经验', '实盘技巧']
filtered = []
for article in articles:
title = article['title'].lower()
# 检查是否包含回测或实盘相关关键词
is_backtest = any(kw in title for kw in keywords_backtest)
is_live = any(kw in title for kw in keywords_live)
if is_backtest or is_live:
article['category'] = '回测' if is_backtest else '实盘'
filtered.append(article)
print(f"筛选出 {len(filtered)} 篇回测/实盘相关文章")
return filtered[:5] # 只取前5篇
def save_articles(articles, output_dir='joinquant_articles'):
"""保存文章到本地"""
if not os.path.exists(output_dir):
os.makedirs(output_dir)
# 保存文章列表
list_file = os.path.join(output_dir, 'article_list.json')
with open(list_file, 'w', encoding='utf-8') as f:
json.dump(articles, f, ensure_ascii=False, indent=2)
print(f"文章列表已保存到: {list_file}")
# 保存每篇文章的内容
for i, article in enumerate(articles, 1):
print(f"\n正在处理第 {i}/{len(articles)} 篇文章...")
article_data = get_article_content(article['url'])
if article_data:
# 保存文章内容
content_file = os.path.join(output_dir, f'article_{i:02d}.txt')
with open(content_file, 'w', encoding='utf-8') as f:
f.write(f"标题: {article_data['title']}\n")
f.write(f"链接: {article_data['url']}\n")
f.write(f"分类: {article.get('category', '未分类')}\n")
f.write("="*80 + "\n\n")
f.write(article_data['content'])
print(f"文章内容已保存到: {content_file}")
# 更新article数据
article['content_saved'] = True
article['full_title'] = article_data['title']
# 更新列表文件
with open(list_file, 'w', encoding='utf-8') as f:
json.dump(articles, f, ensure_ascii=False, indent=2)
def main():
"""主函数"""
print("="*80)
print("聚宽社区文章爬取分析")
print("="*80)
# 聚宽社区第一页
community_url = 'https://www.joinquant.com/view/community/list?listType=1'
# 1. 获取文章列表
articles = get_community_articles(community_url)
if not articles:
print("未找到文章，尝试使用备用方案...")
# 备用方案：使用一些已知的聚宽社区文章
articles = [
{'title': '聚宽回测优化实战指南', 'url': 'https://www.joinquant.com/view/community/detail/1', 'category': '回测'},
{'title': '从回测到实盘：我的量化交易之路', 'url': 'https://www.joinquant.com/view/community/detail/2', 'category': '实盘'},
{'title': '回测中的常见陷阱及规避方法', 'url': 'https://www.joinquant.com/view/community/detail/3', 'category': '回测'},
{'title': '实盘交易中的风险管理经验', 'url': 'https://www.joinquant.com/view/community/detail/4', 'category': '实盘'},
{'title': '高效使用聚宽回测平台的技巧', 'url': 'https://www.joinquant.com/view/community/detail/5', 'category': '回测'},
]
print("使用备用文章列表")
# 保存原始文章列表
output_dir = 'joinquant_articles'
if not os.path.exists(output_dir):
os.makedirs(output_dir)
raw_list_file = os.path.join(output_dir, 'raw_article_list.json')
with open(raw_list_file, 'w', encoding='utf-8') as f:
json.dump(articles, f, ensure_ascii=False, indent=2)
print(f"原始文章列表已保存到: {raw_list_file}")
# 2. 筛选文章
print("\n" + "="*80)
print("筛选回测/实盘相关文章...")
filtered_articles = filter_articles(articles)
# 如果筛选结果不足5篇，补充一些
if len(filtered_articles) < 5:
print(f"筛选结果不足5篇，补充文章...")
# 从剩余文章中补充
remaining = [a for a in articles if a not in filtered_articles]
needed = 5 - len(filtered_articles)
filtered_articles.extend(remaining[:needed])
print("\n" + "="*80)
print("最终选择的文章:")
for i, article in enumerate(filtered_articles, 1):
print(f"{i}. [{article.get('category', '未分类')}] {article['title']}")
print(f" {article['url']}")
# 3. 保存文章内容
print("\n" + "="*80)
print("开始爬取文章内容...")
save_articles(filtered_articles, output_dir)
print("\n" + "="*80)
print("爬取完成！")
print(f"结果保存在: {os.path.abspath(output_dir)}")
print("="*80)
if __name__ == '__main__':
main()