# jq-to-ptrade **Repository Path**: linbirg/jq-to-ptrade ## Basic Information - **Project Name**: jq-to-ptrade - **Description**: jq-to-ptrade skill - **Primary Language**: Unknown - **License**: Not specified - **Default Branch**: master - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 1 - **Forks**: 1 - **Created**: 2026-05-10 - **Last Updated**: 2026-05-12 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # jq-to-ptrade：将聚宽策略迁移至Ptrade的开源工具箱 > 一套将 JoinQuant（聚宽）量化策略一键迁移至 Ptrade 平台的完整方案。包含 Skill 文档、财务 API 全集映射、以及多策略迁移实录。 --- ## 零、项目简介 `jq-to-ptrade` 是一个帮助量化开发者将 JoinQuant 平台策略迁移至 Ptrade（Ptrade 量化交易平台）的开源技能工具箱。 **核心能力：** - 将 JQ 策略代码重写为 Ptrade 可运行的单文件 Python 策略 - 完整保留 JQ 的 `pd.Panel` 多季度财务数据交集逻辑 - 包含 JQ → Ptrade 全集字段映射表 - 提供内联适配层，零外部依赖 **目录结构：** ``` jq-to-ptrade/ ├── SKILL.md # 技能主文档（迁移流程） ├── README.md # 本文：完整使用说明 + 技术干货 ├── references/ │ ├── ptrade-financial-api.md # Ptrade 财务 API 官方文档 │ └── jq-to-ptrade-api-mapping.md # JQ → Ptrade 字段全集映射 └── docs/ ├── 2026-05-04-design.md # jq_real_rsrs_val 迁移设计 └── 2026-05-10-panel-fix.md # Panel 缓存逻辑修复 ``` --- ## 一、为什么做这个 聚宽（JoinQuant）是国内最成熟的量化研究平台之一，拥有丰富的策略资源。但实盘用户往往需要在 Ptrade 等交易平台运行策略时，发现两者 API 并不兼容——JQ 有完整的 `query ORM` + `pd.Panel`，Ptrade 只有分表的 `get_fundamentals`。 **迁移的核心挑战：** | 挑战 | JQ 提供 | Ptrade 提供 | |------|---------|-------------| | 多季度财务交集 | `pd.Panel` 四季度自动切片 | 无 Panel，需手动分季度查 | | 财务字段命名 | `indicator.eps` | `growth_ability.diluted_eps_yoy` | | ST/停牌过滤 | `get_current_data()[s].is_st` | `get_stock_status(list, query_type='ST')` | | 持仓属性 | `position.price` | `position.last_sale_price` | | 日志级别 | `log.set_level` | 不支持，直接删除 | 本工具箱将上述问题逐一解决，并提供完整可运行的 Skill 文档和 API 映射手册。 --- ## 二、迁移流程概览 ``` Step 1: 解析 JQ 专有 API ↓ Step 2: 构建内联适配层（jq_attr_history / jq_filter_st_stock 等） ↓ Step 3: 重写 get_fundamentals（分表查 + pandas join） ↓ Step 4: 重建 Panel 多季度交集 + g_panel 缓存 ↓ Step 5: 修复 position 属性 / 时间参数 ↓ Step 6: 语法检查 + 回测调通 ``` --- ## 三、完整迁移实录：以 jq_real_rsrs_val 为例 > jq_real_rsrs_val 是一个"RSRS 择时 + 价值选股"策略，约 2400 行代码，包含多季度财务过滤逻辑。本节详细展示每个迁移步骤。 ### 3.1 策略结构分析 原始 JQ 策略模块： | 模块 | 功能 | 迁移处理 | |------|------|----------| | `ValueLib.filter_stocks_for_buy` | 多季度财务选股 | **重写**：Panel → 分季度查 | | `RSRSLib` | RSRS 择时（statsmodels OLS） | **保留**：`statsmodels.api as sm` | | `StopManager` | 移动止损 | **保留** | | `TradeStrategyHL` | 涨100%清仓 | **保留** | | `BzUtil` | ST过滤/名称获取 | **内联适配** | | `get_fundamentals` | 财务查询 | **重写**：分表+字段映射 | --- ### 3.2 Step 2：内联适配层 将 JQ 专有 API 封装为 Ptrade 等效函数，写入策略文件顶部： ```python # JQ attribute_history 等效 def jq_attr_history(security, count, unit, fields): freq_map = {'1d': '1d', '1m': '1m', '5m': '5m', '15m': '15m', '30m': '30m', '60m': '60m'} frequency = freq_map.get(unit, unit) return get_history(count, frequency, fields, security, fq='pre', include=True) # JQ get_security_info().display_name 等效（返回值是 dict） def jq_stock_name(security): name_map = get_stock_name(security) if isinstance(name_map, dict): return name_map.get(security, security) return name_map # ST 过滤（JQ: get_current_data()[s].is_st） def jq_filter_st_stock(stock_list): if not stock_list: return [] status_map = get_stock_status(stock_list, query_type='ST') or {} name_map = get_stock_name(stock_list) result = [] for stock in stock_list: name = name_map.get(stock, stock) if isinstance(name, dict): name = name.get(stock, stock) if 'ST' in str(name) or '*' in str(name) or '退' in str(name): continue if status_map.get(stock, False): continue result.append(stock) return result # 停牌过滤（JQ: get_current_data()[s].paused） def jq_filter_paused_stock(stock_list): if not stock_list: return [] status_map = get_stock_status(stock_list, query_type='HALT') or {} return [s for s in stock_list if not status_map.get(s, False)] ``` --- ### 3.3 Step 4（核心难点）：重建 Panel 多季度交集逻辑 JQ 使用 `pd.Panel` 存储近 4 个季度的多指标数据，每个 filter 方法遍历 4 个季度取交集——**股票必须连续 4 个季度都满足条件才入选**。 **JQ 原版 filter_by_4q_eps_between：** ```python def filter_by_4q_eps_between(cls, stocks, panel, area=(0.08, 0.8)): l6 = set() for i in range(4): df_6 = panel.iloc[:, i, :] # 取第 i 季度全部股票 df_temp = df_6[(df_6['eps'] > area[0]) & (df_6['eps'] < area[1])] if i == 0: l6 = set(df_temp.index) # 第一季度满足条件股票 if i > 0: l6 = l6 & set(df_temp.index) # 与前几季度取交集 return [s for s in stocks if s in l6] ``` **Ptrade 重建方案：** ```python # 模块级 Panel 缓存（模拟 JQ g.panel） g_panel = {} def _get_panel_key(current_date): return str(current_date)[:10] def _get_cached_panel(current_date): key = _get_panel_key(current_date) if key not in g_panel: g_panel[key] = ValueLib.get_quarter_fundamentals(BzUtil.get_all_stocks(), 4) return g_panel[key] @classmethod def get_quarter_fundamentals(cls, stocks, num=4): # 按当前季度统一计算 quarters = [当前季度, q1, q2, q3] current_date = getattr(g, 'current_dt', None) current_date_str = str(current_date) year, month = int(current_date_str[0:4]), int(current_date_str[5:7]) quarter_map = {12: 'q4', 9: 'q3', 6: 'q2', 3: 'q1'} current_q = quarter_map.get(month, 'q1') quarters = [current_q] for _ in range(num - 1): quarters.append(cls.get_pre_quarter(quarters[-1])) panel_dict = {} # {quarter_idx: {stock: {field: value}}} for q_idx, quarter in enumerate(quarters): q_data = {} pro_df = get_fundamentals(stocks, 'profit_ability', ['roe', 'gross_income_ratio'], statDate=quarter) cfs_df = get_fundamentals(stocks, 'cashflow_statement', ['net_operate_cash_flow', 'net_invest_cash_flow'], statDate=quarter) bal_df = get_fundamentals(stocks, 'balance_statement', ['total_current_assets', 'total_current_liability'], statDate=quarter) gro_df = get_fundamentals(stocks, 'growth_ability', ['diluted_eps_yoy', 'operating_revenue_grow_rate'], statDate=quarter) for df in [pro_df, cfs_df, bal_df, gro_df]: if df is None or len(df) == 0: continue df = df.set_index('code') if 'code' in df.columns else df for stock in df.index: if stock not in q_data: q_data[stock] = {} for col in df.columns: if col not in ('publ_date', 'end_date', 'company_type', 'secu_abbr'): q_data[stock][col] = df.at[stock, col] if col in df.columns else 0 panel_dict[q_idx] = q_data g_panel[_get_panel_key(current_date)] = panel_dict return panel_dict ``` **Ptrade 重写后的 filter_by_4q_eps_between：** ```python @classmethod def filter_by_4q_eps_between(cls, stocks, current_date, area=(0.08, 0.8)): panel_data = _get_cached_panel(current_date) l6 = set() for i in range(4): q_data = panel_data.get(i, {}) if i == 0: l6 = set([ s for s in stocks if s in q_data and area[0] < q_data[s].get('diluted_eps_yoy', 0) < area[1] ]) else: l_temp = set([ s for s in stocks if s in q_data and area[0] < q_data[s].get('diluted_eps_yoy', 0) < area[1] ]) l6 = l6 & l_temp # 四季度交集 log.info("近四季盈余成长率介于%d至%d:%d" % (area[0] * 100, area[1] * 100, len(l6))) return list(l6) ``` **日终清空缓存：** ```python def after_market_close(context): global g_panel g_panel = {} # 下个交易日重新构建 ``` --- ### 3.4 财务字段映射（关键） JQ 的 `eps` 在 Ptrade 中对应 `growth_ability.diluted_eps_yoy`（稀释每股收益同比增长），**不是** `eps.eps`（每股收益的绝对值）。 | JQ 字段 | Ptrade 表 | Ptrade 字段 | 含义 | |---------|-----------|-------------|------| | `indicator.eps` | growth_ability | `diluted_eps_yoy` | 每股收益增长率（%） | | `indicator.inc_revenue_year_on_year` | growth_ability | `operating_revenue_grow_rate` | 营收同比增长（%） | | `indicator.roe` | profit_ability | `roe` | 净资产收益率（%） | | `indicator.gross_margin` | profit_ability | `gross_income_ratio` | 销售毛利率（%） | | `cash_flow.net_operate_cash_flow` | cashflow_statement | `net_operate_cash_flow` | 经营活动现金流 | | `cash_flow.net_invest_cash_flow` | cashflow_statement | `net_invest_cash_flow` | 投资活动现金流 | | `balance.total_current_assets` | balance_statement | `total_current_assets` | 流动资产 | | `balance.total_current_liability` | balance_statement | `total_current_liability` | 流动负债 | | `income.operating_revenue` | income_statement | `operating_revenue` | 营业收入 | 完整映射表见 `references/jq-to-ptrade-api-mapping.md`。 --- ### 3.5 Step 5：Position 属性修复 Ptrade 持仓对象属性名与 JQ 不同： | JQ | Ptrade | |----|--------| | `position.price` | `position.last_sale_price` | | `position.avg_cost` | `position.cost_basis` | | `position.security` | `position.sid` | | `position.total_amount` | `position.amount` | | `position.value` | `position.last_sale_price * position.amount` | --- ### 3.6 Step 6：常见 API 坑点 | 问题 | JQ | Ptrade 处理 | |------|----|-----------| | `time='open'` | 支持 | 不支持，改为 `'9:40'` | | `time='after_close'` | 支持 | 不支持，改为 `'15:00'` | | `log.set_level` | 支持 | 不支持，删除 | | `log.warn` | 支持 | Ptrade 用 `log.warning` | | `set_slippage(PriceRelatedSlippage(...))` | 对象参数 | Ptrade 直接接受数值 | | `get_price(end_date=datetime.date)` | datetime对象 | 必须是字符串 `'YYYY-MM-DD'` | | `get_price` 返回 `None` | 不会 | 需先做空值检查 | --- ## 四、快速上手 ### 4.1 安装 Skill 将本仓库 `SKILL.md` 路径添加到 opencode 配置，或直接将 `skills/` 目录放入 opencode skills 路径： ```bash # 或通过 symlink ln -s /path/to/jq-to-ptrade ~/.config/opencode/skills/jq-to-ptrade ``` ### 4.2 使用流程 ``` 1. 准备好 JQ 策略源码（.py 文件） 2. 引用 jq-to-ptrade skill 3. 用 brainstorming 确认迁移范围 4. 按 SKILL.md Step 1~7 执行迁移 5. python -m py_compile 验证语法 6. 在 Ptrade 回测中调通 ``` ### 4.3 文件清单 | 文件 | 说明 | |------|------| | `SKILL.md` | 完整迁移流程（Skill 格式） | | `references/ptrade-financial-api.md` | Ptrade 财务 API 官方完整字段说明 | | `references/jq-to-ptrade-api-mapping.md` | JQ → Ptrade 字段全集映射（9张表，200+字段） | | `docs/2026-05-04-design.md` | jq_real_rsrs_val 迁移设计文档 | | `docs/2026-05-10-panel-fix.md` | Panel 缓存逻辑修复设计文档 | --- ## 五、已知限制 1. **`get_ticks`**：Ptrade 无逐笔 tick 接口，依赖此 API 的策略无法迁移 2. **`statsmodels` 依赖**：`RSRSLib` 依赖 `statsmodels.api as sm`，需在 Ptrade 策略中保留 3. **Panel 缓存**：`g_panel` 为模块级变量，日终清空；若 Ptrade 策略运行于多进程模式需自行改造 4. **字段实测**：部分字段（如 `gross_income_ratio`、`diluted_eps_yoy`）需在实盘前实测确认数据质量 --- ## 六、字段映射速查 完整映射表见 `references/jq-to-ptrade-api-mapping.md`，以下是高频使用字段： ``` valuation → total_value, float_value, pe_ttm, pb profit_ability → roe, roa, gross_income_ratio growth_ability → diluted_eps_yoy, operating_revenue_grow_rate, np_parent_company_yoy cashflow_stmt → net_operate_cash_flow, net_invest_cash_flow balance_stmt → total_current_assets, total_current_liability, good_will income_stmt → operating_revenue, net_profit eps → basic_eps, diluted_eps, naps operating → inventory_turnover_rate, accounts_receivables_turnover_days debt_paying → current_ratio, quick_ratio, debt_equity_ratio ``` --- ## 七、许可与联系 MIT License。欢迎 Star 和 Pull Request。 如有问题或发现新的 API 差异，请提交 Issue。