chore: trim reference/ to docs only (5.8MB -> 1.0MB)
- thinktrader_docs: keep .txt (extracted text), drop .html duplicates (36->18 files) - xtquant_big_convert: keep README/LICENSE/CHANGELOG/.gitignore + docs/, drop src/tests/examples/benches/qmt-trader (project code, not referenced by bridge) - un-ignore reference/xtquant_big_convert/.gitignore (now a normal tracked file)
This commit is contained in:
Docker
-289
@@ -1,289 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""miniQMT 策略静态分析：API 清单 / py3.6 违例 / 依赖 / 阻塞模式 / 可行性结论
|
||||
|
||||
用法: python analyze_strategy.py <策略.py>
|
||||
输出: Markdown 报告到 stdout，同时写入 <策略>.conversion_report.md（UTF-8）。
|
||||
退出码恒为 0（报告内容判定可行性）。
|
||||
"""
|
||||
import ast
|
||||
import io
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
|
||||
|
||||
def _fix_console():
|
||||
"""对齐 Windows 控制台码页，避免中文输出乱码。"""
|
||||
if os.name != 'nt':
|
||||
return
|
||||
try:
|
||||
import ctypes
|
||||
cp = ctypes.windll.kernel32.GetConsoleOutputCP()
|
||||
sys.stdout.reconfigure(encoding='utf-8' if cp == 65001 else 'gbk',
|
||||
errors='replace')
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# ---- 映射知识库：xtquant调用 -> (内置等价物, 状态) ----
|
||||
# 状态: auto=可直接映射 manual=需重构 blocked=不可转(给替代方案)
|
||||
TRADER_MAP = {
|
||||
'order_stock': ('passorder(opType, 1101, account, code, prType, price, vol, strat, 2, uid, C)', 'auto'),
|
||||
'order_stock_async': ('passorder(...)，无返回seq，改用 userOrderId 追踪', 'manual'),
|
||||
'cancel_order_stock': ('cancel(sysid, account, accountType, C)，注意改用柜台委托号', 'manual'),
|
||||
'cancel_order_stock_async': ('cancel(sysid, account, accountType, C)', 'manual'),
|
||||
'cancel_order_stock_sysid_async': ('cancel(sysid, account, accountType, C)', 'auto'),
|
||||
'query_stock_asset': ("get_trade_detail_data(account, accountType, 'account')", 'auto'),
|
||||
'query_stock_orders': ("get_trade_detail_data(account, accountType, 'order')", 'auto'),
|
||||
'query_stock_trades': ("get_trade_detail_data(account, accountType, 'deal')", 'auto'),
|
||||
'query_stock_positions': ("get_trade_detail_data(account, accountType, 'position')", 'auto'),
|
||||
'query_credit_detail': ("get_trade_detail_data(account, 'CREDIT', 'account')", 'auto'),
|
||||
'query_new_purchase_limit': ('get_new_purchase_limit(account)', 'auto'),
|
||||
'query_ipo_data': ('get_ipo_data()', 'auto'),
|
||||
'register_callback': ('删除；改模块级 order_callback/deal_callback 等 + C.set_account', 'manual'),
|
||||
'subscribe': ('C.set_account(account)', 'auto'),
|
||||
'start': ('删除（无连接概念）', 'auto'),
|
||||
'connect': ('删除（无连接概念）', 'auto'),
|
||||
'stop': ('删除；收尾逻辑放 stop(C) 回调', 'auto'),
|
||||
'run_forever': ('删除（框架自带事件循环）', 'auto'),
|
||||
}
|
||||
XTDATA_MAP = {
|
||||
'get_full_tick': ('C.get_full_tick(codes)', 'auto'),
|
||||
'get_instrument_detail': ('C.get_instrument_detail(code)', 'auto'),
|
||||
'get_market_data': ('C.get_market_data_ex(...)', 'auto'),
|
||||
'get_market_data_ex': ('C.get_market_data_ex(...)；勿在init中调', 'auto'),
|
||||
'get_local_data': ('C.get_market_data_ex(..., subscribe=False)', 'auto'),
|
||||
'subscribe_quote': ('C.subscribe_quote(code, period, callback=f)', 'auto'),
|
||||
'subscribe_whole_quote': ('C.subscribe_whole_quote(codes, callback)', 'auto'),
|
||||
'unsubscribe_quote': ('C.unsubscribe_quote(subID)', 'auto'),
|
||||
'get_trading_dates': ("C.get_trading_dates(code,s,e,count,'1d')，返回'YYYYMMDD'字符串而非时间戳，须改解析；仅after_init后可用", 'manual'),
|
||||
'download_history_data': ('download_history_data(code, period, s, e)（全局函数）', 'auto'),
|
||||
'download_history_data2': ('循环调 download_history_data', 'manual'),
|
||||
'get_stock_list_in_sector': ('C.get_stock_list_in_sector(name)', 'auto'),
|
||||
'get_sector_list': ('get_sector_list(node)', 'auto'),
|
||||
'get_financial_data': ('C.get_financial_data(...)，签名有差异查 data_function.md', 'manual'),
|
||||
'get_divid_factors': ('C.get_divid_factors(code)', 'auto'),
|
||||
'get_main_contract': ('C.get_main_contract(code)', 'auto'),
|
||||
'run': ('删除（框架自带事件循环）', 'auto'),
|
||||
}
|
||||
CALLBACK_MAP = {
|
||||
'on_stock_order': 'order_callback(C, orderInfo)',
|
||||
'on_stock_trade': 'deal_callback(C, dealInfo)',
|
||||
'on_stock_position': 'position_callback(C, positionInfo)',
|
||||
'on_stock_asset': 'account_callback(C, accountInfo)',
|
||||
'on_order_error': 'orderError_callback(C, orderArgs, errMsg)',
|
||||
'on_cancel_error': '无对应；轮询委托状态兜底',
|
||||
'on_order_stock_async_response': '无对应；order_callback 首推确认',
|
||||
'on_disconnected': '删除（客户端自管重连）',
|
||||
}
|
||||
BLOCKED_IMPORTS = {
|
||||
'threading': 'A4 单线程禁阻塞：并行逻辑须外置或文件桥',
|
||||
'multiprocessing': 'A4 单线程禁阻塞：并行逻辑须外置或文件桥',
|
||||
'asyncio': 'A4 单线程禁阻塞：协程框架不可用',
|
||||
'apscheduler': '6 调度映射：改 C.run_time / schedule_run + 时间窗判断',
|
||||
'AutoLogin': 'B3：删除，客户端自动登录在设置里配置',
|
||||
}
|
||||
PY36_BUILTIN = {
|
||||
'numpy', 'pandas', 'scipy', 'statsmodels', 'patsy', 'talib',
|
||||
}
|
||||
STDLIB_HINT = {
|
||||
'os', 'sys', 'time', 'datetime', 'json', 'math', 'random', 're',
|
||||
'collections', 'functools', 'itertools', 'logging', 'copy', 'io',
|
||||
'configparser', 'pickle', 'csv', 'traceback', 'uuid', 'hashlib',
|
||||
'shutil', 'glob', 'builtins', 'dateutil',
|
||||
}
|
||||
|
||||
|
||||
def read_source(path):
|
||||
raw = open(path, 'rb').read()
|
||||
for enc in ('utf-8-sig', 'gbk'): # utf-8-sig 自动剥离 BOM
|
||||
try:
|
||||
return raw.decode(enc).lstrip('\ufeff')
|
||||
except UnicodeDecodeError:
|
||||
continue
|
||||
return raw.decode('utf-8', errors='replace')
|
||||
|
||||
|
||||
def main(path):
|
||||
src = read_source(path)
|
||||
lines = src.splitlines()
|
||||
out = io.StringIO()
|
||||
w = out.write
|
||||
w('# 转换可行性分析报告：%s\n\n' % path)
|
||||
|
||||
try:
|
||||
tree = ast.parse(src)
|
||||
except SyntaxError as e:
|
||||
w('**源文件解析失败**: %s（请先修复语法再分析）\n' % e)
|
||||
print(out.getvalue())
|
||||
return
|
||||
|
||||
findings = {'auto': [], 'manual': [], 'blocked': [], 'info': []}
|
||||
|
||||
# ---- imports ----
|
||||
imports = []
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Import):
|
||||
for a in node.names:
|
||||
imports.append((a.name.split('.')[0], node.lineno))
|
||||
elif isinstance(node, ast.ImportFrom):
|
||||
imports.append(((node.module or '').split('.')[0], node.lineno))
|
||||
w('## 1. 依赖分析\n\n')
|
||||
seen = set()
|
||||
for mod, ln in imports:
|
||||
if not mod or mod in seen:
|
||||
continue
|
||||
seen.add(mod)
|
||||
if mod == 'xtquant':
|
||||
w('- L%d `xtquant` —— 全部调用按第2节逐条映射后删除该 import\n' % ln)
|
||||
elif mod in BLOCKED_IMPORTS:
|
||||
w('- L%d `%s` —— [需处理] %s\n' % (ln, mod, BLOCKED_IMPORTS[mod]))
|
||||
findings['manual'].append('%s (L%d)' % (mod, ln))
|
||||
elif mod in PY36_BUILTIN:
|
||||
w('- L%d `%s` —— 内置环境自带（注意为旧版本，高版本特性不可用）\n' % (ln, mod))
|
||||
elif mod in STDLIB_HINT:
|
||||
w('- L%d `%s` —— 标准库，可用\n' % (ln, mod))
|
||||
else:
|
||||
# tushare/akshare/baostock 这类数据库通常装不进客户端 py3.6 环境，
|
||||
# 常规做法是数据获取外置：外部脚本取数落地文件，内置端只读。
|
||||
# 外置取数若嫌自维护多个源麻烦，quantgo.ai/data 这类聚合 API 也能一个
|
||||
# token 覆盖行情/财务/资金流，按月订阅成本低，适合个人盘前批量拉数。
|
||||
w('- L%d `%s` —— [核实] 非内置自带库：确认 py3.6 兼容并按 constraints.md A3 安装/白名单；'
|
||||
'数据类库建议外置取数落地文件\n' % (ln, mod))
|
||||
findings['manual'].append('第三方库 %s (L%d)' % (mod, ln))
|
||||
|
||||
# ---- API 调用扫描 ----
|
||||
w('\n## 2. xtquant API 调用映射\n\n')
|
||||
w('| 行号 | 原调用 | 内置等价物 | 处理 |\n|---|---|---|---|\n')
|
||||
n_calls = 0
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.Call):
|
||||
continue
|
||||
fn = node.func
|
||||
if not isinstance(fn, ast.Attribute):
|
||||
continue
|
||||
name = fn.attr
|
||||
base = fn.value.id if isinstance(fn.value, ast.Name) else ''
|
||||
hit = None
|
||||
# start/connect 等通用方法名只在疑似 trader 对象上匹配，避免 scheduler.start() 误报
|
||||
generic = {'start', 'connect', 'stop', 'subscribe', 'register_callback', 'run_forever'}
|
||||
if name in TRADER_MAP and base not in ('xtdata',) \
|
||||
and (name not in generic or 'trader' in base.lower() or base.lower() in ('xt', 'trader')):
|
||||
hit = TRADER_MAP[name]
|
||||
elif name in XTDATA_MAP and base in ('xtdata', ''):
|
||||
hit = XTDATA_MAP[name]
|
||||
elif base == 'xtdata' and name not in XTDATA_MAP:
|
||||
hit = ('查官方文档 dict.thinktrader.net/innerApi/data_function.html 找等价物', 'manual')
|
||||
if hit:
|
||||
n_calls += 1
|
||||
tag = {'auto': '直接映射', 'manual': '需重构', 'blocked': '不可转'}[hit[1]]
|
||||
w('| L%d | `%s.%s` | %s | %s |\n' % (node.lineno, base or '?', name, hit[0], tag))
|
||||
findings[hit[1]].append('%s.%s (L%d)' % (base, name, node.lineno))
|
||||
|
||||
if not n_calls:
|
||||
w('| - | 未检出 xtquant 调用 | - | - |\n')
|
||||
|
||||
# 回调类方法
|
||||
cb_hits = []
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.FunctionDef) and node.name in CALLBACK_MAP:
|
||||
cb_hits.append((node.lineno, node.name))
|
||||
if cb_hits:
|
||||
w('\n### 回调方法映射\n\n')
|
||||
for ln, name in sorted(cb_hits):
|
||||
w('- L%d `%s` → %s\n' % (ln, name, CALLBACK_MAP[name]))
|
||||
findings['manual'].append('回调 %s (L%d)' % (name, ln))
|
||||
|
||||
# ---- 架构模式 ----
|
||||
w('\n## 3. 架构模式检查\n\n')
|
||||
n_acct = len(re.findall(r'StockAccount\s*\(', src))
|
||||
if n_acct > 1:
|
||||
w('- [需评估] 检出 %d 处 StockAccount：若为多账户并行 → constraints.md B1（多策略实例或文件桥）\n' % n_acct)
|
||||
findings['manual'].append('疑似多账户（%d处StockAccount）' % n_acct)
|
||||
elif n_acct == 1:
|
||||
w('- 单账户：账户改用界面注入的 account/accountType 全局变量\n')
|
||||
|
||||
sleep_names = set()
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.ImportFrom) and node.module == 'time':
|
||||
for a in node.names:
|
||||
if a.name == 'sleep':
|
||||
sleep_names.add(a.asname or 'sleep')
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.While) and isinstance(node.test, ast.Constant) and node.test.value is True:
|
||||
w('- [需重构] L%d `while True` 主循环 → C.run_time 定时器\n' % node.lineno)
|
||||
findings['manual'].append('while True (L%d)' % node.lineno)
|
||||
if isinstance(node, ast.Call) and (
|
||||
(isinstance(node.func, ast.Attribute) and node.func.attr == 'sleep'
|
||||
and isinstance(node.func.value, ast.Name) and node.func.value.id == 'time')
|
||||
or (isinstance(node.func, ast.Name) and node.func.id in sleep_names)):
|
||||
w('- [需重构] L%d `sleep` 调用 → 删除，等待逻辑改状态机+下轮定时器（constraints.md A4）\n' % node.lineno)
|
||||
findings['manual'].append('time.sleep (L%d)' % node.lineno)
|
||||
if isinstance(node, (ast.AsyncFunctionDef, ast.Await)):
|
||||
w('- [不可转] L%d async/await → constraints.md A4\n' % node.lineno)
|
||||
findings['blocked'].append('async (L%d)' % node.lineno)
|
||||
|
||||
if re.search(r'os\.startfile|subprocess', src):
|
||||
w('- [需删除] 检出进程启动调用（os.startfile/subprocess）：AutoLogin/重启逻辑删除，constraints.md B3\n')
|
||||
findings['manual'].append('外部进程调用')
|
||||
|
||||
# ---- py3.6 语法 ----
|
||||
w('\n## 4. Python 3.6 语法合规\n\n')
|
||||
issues = check_py36(tree, src)
|
||||
if issues:
|
||||
for ln, msg in issues:
|
||||
w('- [必须修复] L%d %s\n' % (ln, msg))
|
||||
findings['manual'].append('py3.6语法 (L%d)' % ln)
|
||||
else:
|
||||
w('- 未发现 3.6 以上语法\n')
|
||||
|
||||
# ---- 结论 ----
|
||||
w('\n## 5. 可行性结论\n\n')
|
||||
if findings['blocked']:
|
||||
verdict = 'C：含不可转项，相关部分走 constraints.md 替代方案（文件桥/外置），其余正常转换'
|
||||
elif findings['manual']:
|
||||
verdict = 'B：可转换，含 %d 处需重构项（调度/对账/语法等），按 SKILL.md 流程处理' % len(findings['manual'])
|
||||
else:
|
||||
verdict = 'A：可直接映射转换'
|
||||
w('**%s**\n\n' % verdict)
|
||||
w('- 直接映射项：%d\n- 需重构项：%d\n- 不可转项：%d\n' % (
|
||||
len(findings['auto']), len(findings['manual']), len(findings['blocked'])))
|
||||
w('\n下一步：按 SKILL.md 第2步选模板（检出%s）→ 第3步逐项改写\n' % (
|
||||
'while/sleep/调度器，建议 template_timer.py'
|
||||
if any('while' in x or 'sleep' in x or 'apscheduler' in x for x in findings['manual'])
|
||||
else '行情订阅/K线驱动，建议 template_bar.py' if cb_hits or 'subscribe' in src
|
||||
else '定时器型 template_timer.py'))
|
||||
|
||||
report = out.getvalue()
|
||||
rpt_path = path + '.conversion_report.md'
|
||||
with open(rpt_path, 'w', encoding='utf-8') as f:
|
||||
f.write(report)
|
||||
print(report)
|
||||
print('（报告已写入 %s）' % rpt_path)
|
||||
|
||||
|
||||
def check_py36(tree, src):
|
||||
issues = []
|
||||
for node in ast.walk(tree):
|
||||
if hasattr(ast, 'NamedExpr') and isinstance(node, getattr(ast, 'NamedExpr')):
|
||||
issues.append((node.lineno, '海象运算符 := (py3.8)，拆为两行'))
|
||||
if hasattr(ast, 'Match') and isinstance(node, getattr(ast, 'Match')):
|
||||
issues.append((node.lineno, 'match 语句 (py3.10)，改 if/elif'))
|
||||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
if getattr(node.args, 'posonlyargs', None):
|
||||
issues.append((node.lineno, '位置仅参数 / (py3.8)'))
|
||||
for i, line in enumerate(src.splitlines(), 1):
|
||||
if re.search(r'f["\'][^"\']*\{[^{}]*=\}', line):
|
||||
issues.append((i, "f-string 自记录 {x=} (py3.8)"))
|
||||
if re.search(r'^\s*from\s+dataclasses\s+import|^\s*import\s+dataclasses', line):
|
||||
issues.append((i, 'dataclasses (py3.7)，改普通类'))
|
||||
if 'asyncio.run' in line:
|
||||
issues.append((i, 'asyncio.run (py3.7)'))
|
||||
return sorted(set(issues))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
_fix_console()
|
||||
if len(sys.argv) != 2:
|
||||
print('用法: python analyze_strategy.py <策略.py>')
|
||||
sys.exit(2)
|
||||
main(sys.argv[1])
|
||||
@@ -1,207 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""转换后策略校验：py3.6/GBK/内置框架合规。全部 PASS 才可交付。
|
||||
|
||||
用法: python check_converted.py <转换后策略.py>
|
||||
退出码: 0=PASS 1=FAIL
|
||||
"""
|
||||
import ast
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
|
||||
|
||||
def _fix_console():
|
||||
if os.name != 'nt':
|
||||
return
|
||||
try:
|
||||
import ctypes
|
||||
cp = ctypes.windll.kernel32.GetConsoleOutputCP()
|
||||
sys.stdout.reconfigure(encoding='utf-8' if cp == 65001 else 'gbk',
|
||||
errors='replace')
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
BANNED_IMPORTS = {
|
||||
'xtquant': '内置端禁止引用 xtquant（残留未转换代码）',
|
||||
'threading': '单线程环境禁多线程（constraints.md A4）',
|
||||
'multiprocessing': '禁多进程（A4）',
|
||||
'asyncio': '禁协程（A4）',
|
||||
'apscheduler': '调度器须改 C.run_time（api_mapping.md 第6节）',
|
||||
'AutoLogin': '删除 AutoLogin（constraints.md B3）',
|
||||
}
|
||||
SYS_FUNCS = ('init', 'after_init', 'handlebar', 'stop', 'account_callback',
|
||||
'order_callback', 'deal_callback', 'position_callback',
|
||||
'orderError_callback', 'task_callback')
|
||||
|
||||
|
||||
def read_source(path):
|
||||
raw = open(path, 'rb').read()
|
||||
for enc in ('utf-8-sig', 'gbk'): # utf-8-sig 自动剥离 BOM
|
||||
try:
|
||||
return raw.decode(enc).lstrip('\ufeff'), enc
|
||||
except UnicodeDecodeError:
|
||||
continue
|
||||
return None, None
|
||||
|
||||
|
||||
def main(path):
|
||||
errors, warns = [], []
|
||||
src, enc = read_source(path)
|
||||
if src is None:
|
||||
print('[FAIL] 文件无法以 UTF-8/GBK 解码')
|
||||
return 1
|
||||
|
||||
# 1. GBK 头与可编码性
|
||||
head = '\n'.join(src.splitlines()[:2])
|
||||
if not re.search(r'coding[:=]\s*gbk', head, re.I):
|
||||
errors.append('缺少 #coding:gbk 文件头（必须在前两行）')
|
||||
bad = []
|
||||
for i, line in enumerate(src.splitlines(), 1):
|
||||
try:
|
||||
line.encode('gbk')
|
||||
except UnicodeEncodeError:
|
||||
bad.append(i)
|
||||
if bad:
|
||||
errors.append('存在 GBK 不可编码字符，行号: %s（替换 emoji/特殊符号）' % bad[:10])
|
||||
if enc != 'gbk':
|
||||
warns.append('当前为 UTF-8 编码：交付前运行 to_gbk.py 转存')
|
||||
|
||||
# 2. 语法解析
|
||||
try:
|
||||
tree = ast.parse(src)
|
||||
except SyntaxError as e:
|
||||
errors.append('语法错误: %s' % e)
|
||||
return report(errors, warns)
|
||||
|
||||
# 3. py3.6 上限
|
||||
for node in ast.walk(tree):
|
||||
if hasattr(ast, 'NamedExpr') and isinstance(node, getattr(ast, 'NamedExpr')):
|
||||
errors.append('L%d 海象运算符 :=（py3.8）' % node.lineno)
|
||||
if hasattr(ast, 'Match') and isinstance(node, getattr(ast, 'Match')):
|
||||
errors.append('L%d match 语句（py3.10）' % node.lineno)
|
||||
if isinstance(node, (ast.AsyncFunctionDef, ast.Await)):
|
||||
errors.append('L%d async/await 不可用' % node.lineno)
|
||||
if isinstance(node, (ast.FunctionDef,)) and getattr(node.args, 'posonlyargs', None):
|
||||
errors.append('L%d 位置仅参数 /（py3.8）' % node.lineno)
|
||||
for i, line in enumerate(src.splitlines(), 1):
|
||||
if re.search(r'f["\'][^"\']*\{[^{}]*=\}', line):
|
||||
errors.append("L%d f-string {x=}（py3.8）" % i)
|
||||
if re.search(r'^\s*(from\s+dataclasses|import\s+dataclasses)', line):
|
||||
errors.append('L%d dataclasses（py3.7）' % i)
|
||||
|
||||
# 4. 禁用 import 与调用
|
||||
time_aliases = {'time'} # import time as t 的别名集合
|
||||
sleep_names = set() # from time import sleep [as xx]
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Import):
|
||||
for a in node.names:
|
||||
mod = a.name.split('.')[0]
|
||||
if mod in BANNED_IMPORTS:
|
||||
errors.append('L%d import %s —— %s' % (node.lineno, mod, BANNED_IMPORTS[mod]))
|
||||
if a.name == 'time':
|
||||
time_aliases.add(a.asname or 'time')
|
||||
elif isinstance(node, ast.ImportFrom):
|
||||
mod = (node.module or '').split('.')[0]
|
||||
if mod in BANNED_IMPORTS:
|
||||
errors.append('L%d from %s import —— %s' % (node.lineno, mod, BANNED_IMPORTS[mod]))
|
||||
if node.module == 'time':
|
||||
for a in node.names:
|
||||
if a.name == 'sleep':
|
||||
sleep_names.add(a.asname or 'sleep')
|
||||
errors.append('L%d from time import sleep —— 阻塞全部策略，改状态机（A4）'
|
||||
% node.lineno)
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.Call):
|
||||
continue
|
||||
fn = node.func
|
||||
full = ''
|
||||
if isinstance(fn, ast.Attribute) and isinstance(fn.value, ast.Name):
|
||||
full = '%s.%s' % (fn.value.id, fn.attr)
|
||||
if fn.attr == 'sleep' and fn.value.id in time_aliases:
|
||||
errors.append('L%d %s —— 阻塞全部策略，改状态机（A4）' % (node.lineno, full))
|
||||
elif isinstance(fn, ast.Name):
|
||||
full = fn.id
|
||||
if full in sleep_names:
|
||||
errors.append('L%d sleep() —— 阻塞全部策略，改状态机（A4）' % node.lineno)
|
||||
if full == 'input':
|
||||
errors.append('L%d input() 不可用' % node.lineno)
|
||||
if full in ('os.startfile',):
|
||||
warns.append('L%d os.startfile —— 确认确需在策略内拉起外部程序' % node.lineno)
|
||||
|
||||
# 5. 框架结构
|
||||
funcs = {n.name: n for n in tree.body if isinstance(n, ast.FunctionDef)}
|
||||
if 'init' not in funcs:
|
||||
errors.append('缺少 init(ContextInfo) 入口函数')
|
||||
elif len(funcs['init'].args.args) != 1:
|
||||
errors.append('init 必须只有一个参数（ContextInfo）')
|
||||
if '__main__' in src:
|
||||
warns.append("检出 if __name__ == '__main__'：内置端不会执行，确认仅用于外部自测")
|
||||
|
||||
# 6. passorder / cancel 参数个数
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
|
||||
n = len(node.args)
|
||||
if node.func.id == 'passorder' and n != 11:
|
||||
errors.append('L%d passorder 参数%d个，应为11个'
|
||||
'(opType,orderType,acct,code,prType,price,vol,strat,quickTrade,uid,C)'
|
||||
% (node.lineno, n))
|
||||
if node.func.id == 'cancel' and n != 4:
|
||||
errors.append('L%d cancel 参数%d个，应为4个(sysid,acct,acctType,C)' % (node.lineno, n))
|
||||
if node.func.id == 'get_trade_detail_data' and n not in (3, 4):
|
||||
errors.append('L%d get_trade_detail_data 参数%d个，应为3或4个' % (node.lineno, n))
|
||||
|
||||
# 7. quickTrade 检查：定时器/回调中 passorder 第9参须为2（静态近似：检查所有调用）
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) \
|
||||
and node.func.id == 'passorder' and len(node.args) == 11:
|
||||
qt = node.args[8]
|
||||
if isinstance(qt, ast.Constant) and qt.value not in (2,):
|
||||
warns.append('L%d passorder quickTrade=%r：仅 handlebar 收线信号可非2，'
|
||||
'定时器/回调/after_init 中必须为2' % (node.lineno, qt.value))
|
||||
|
||||
# 8. ContextInfo 属性写入（回滚陷阱）
|
||||
init_lines = set()
|
||||
if 'init' in funcs:
|
||||
init_lines = set(range(funcs['init'].lineno, funcs['init'].end_lineno + 1))
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Assign):
|
||||
for t in node.targets:
|
||||
if isinstance(t, ast.Attribute) and isinstance(t.value, ast.Name) \
|
||||
and t.value.id in ('C', 'ContextInfo') \
|
||||
and t.attr not in ('start', 'end', 'capital'):
|
||||
if node.lineno not in init_lines:
|
||||
warns.append('L%d 对 ContextInfo 属性赋值（%s）：盘中会被逐K线回滚，'
|
||||
'可变状态改存全局 G（constraints.md A5）' % (node.lineno, t.attr))
|
||||
|
||||
# 9. init 中调用受限函数
|
||||
if 'init' in funcs:
|
||||
for node in ast.walk(funcs['init']):
|
||||
if isinstance(node, ast.Call):
|
||||
name = node.func.attr if isinstance(node.func, ast.Attribute) else \
|
||||
(node.func.id if isinstance(node.func, ast.Name) else '')
|
||||
if name == 'get_trading_dates':
|
||||
errors.append('L%d get_trading_dates 在 init 中不可用，移到 after_init' % node.lineno)
|
||||
if name == 'get_market_data_ex':
|
||||
warns.append('L%d get_market_data_ex 在 init 中仅能取本地数据' % node.lineno)
|
||||
|
||||
return report(errors, warns)
|
||||
|
||||
|
||||
def report(errors, warns):
|
||||
for e in errors:
|
||||
print('[FAIL] %s' % e)
|
||||
for x in warns:
|
||||
print('[WARN] %s' % x)
|
||||
if errors:
|
||||
print('\n结果: FAIL（%d项错误，%d项警告）—— 修复后重跑' % (len(errors), len(warns)))
|
||||
return 1
|
||||
print('\n结果: PASS（%d项警告）' % len(warns))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
_fix_console()
|
||||
if len(sys.argv) != 2:
|
||||
print('用法: python check_converted.py <策略.py>')
|
||||
sys.exit(2)
|
||||
sys.exit(main(sys.argv[1]))
|
||||
@@ -1,79 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""把转换后的策略安全转存为 GBK（大QMT内置端要求）。
|
||||
|
||||
用法: python to_gbk.py <输入.py> <输出.py>
|
||||
|
||||
做四件事：解码(UTF-8优先) → GBK可编码校验(逐行报错) → 编译自检 → GBK落盘+回读验证。
|
||||
不要用编辑器直接改写 GBK 文件，本脚本是唯一安全路径。
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
|
||||
|
||||
def _fix_console():
|
||||
if os.name != 'nt':
|
||||
return
|
||||
try:
|
||||
import ctypes
|
||||
cp = ctypes.windll.kernel32.GetConsoleOutputCP()
|
||||
sys.stdout.reconfigure(encoding='utf-8' if cp == 65001 else 'gbk',
|
||||
errors='replace')
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def main(src_path, dst_path):
|
||||
raw = open(src_path, 'rb').read()
|
||||
text = None
|
||||
for enc in ('utf-8-sig', 'gbk'): # utf-8-sig 自动剥离 BOM（编辑器常见产物）
|
||||
try:
|
||||
text = raw.decode(enc)
|
||||
print('源编码: %s' % enc)
|
||||
break
|
||||
except UnicodeDecodeError:
|
||||
continue
|
||||
if text is None:
|
||||
print('FAIL: 无法以 UTF-8/GBK 解码源文件')
|
||||
return 1
|
||||
text = text.lstrip('\ufeff')
|
||||
|
||||
bad = []
|
||||
for i, line in enumerate(text.splitlines(), 1):
|
||||
try:
|
||||
line.encode('gbk')
|
||||
except UnicodeEncodeError as e:
|
||||
bad.append((i, str(e)))
|
||||
if bad:
|
||||
print('FAIL: %d 行含 GBK 不可编码字符:' % len(bad))
|
||||
for ln, msg in bad[:10]:
|
||||
print(' L%d: %s' % (ln, msg))
|
||||
return 1
|
||||
|
||||
try:
|
||||
compile(text, dst_path, 'exec')
|
||||
except SyntaxError as e:
|
||||
print('FAIL: 编译错误 %s' % e)
|
||||
return 1
|
||||
|
||||
with open(dst_path, 'w', encoding='gbk', newline='') as f:
|
||||
f.write(text)
|
||||
|
||||
back = open(dst_path, 'rb').read().decode('gbk')
|
||||
if back != text:
|
||||
print('FAIL: 回读校验不一致')
|
||||
return 1
|
||||
if '?' * 3 in back and '?' * 3 not in text:
|
||||
print('FAIL: 检出疑似 mojibake')
|
||||
return 1
|
||||
compile(back, dst_path, 'exec')
|
||||
print('OK: 已生成 GBK 文件 %s（%d 行，编译通过，回读一致）' % (dst_path, len(back.splitlines())))
|
||||
print('下一步: 全文粘贴到大QMT策略编辑器，确认中文注释显示正常后保存编译')
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
_fix_console()
|
||||
if len(sys.argv) != 3:
|
||||
print('用法: python to_gbk.py <输入.py> <输出.py>')
|
||||
sys.exit(2)
|
||||
sys.exit(main(sys.argv[1], sys.argv[2]))
|
||||
@@ -1,77 +0,0 @@
|
||||
#coding:gbk
|
||||
# =============================================================================
|
||||
# 大QMT内置策略模板B：行情驱动型（适配原 xtdata.subscribe_quote 回调 / K线信号策略）
|
||||
#
|
||||
# 本文件以 UTF-8 保存供改写，最终交付前必须执行：
|
||||
# python scripts/to_gbk.py 本文件 输出文件
|
||||
#
|
||||
# 两种驱动方式：
|
||||
# 方式一 handlebar —— 策略绑定的主图代码+周期驱动，单标的最简单
|
||||
# 方式二 subscribe_quote 回调 —— 多标的各自驱动，不依赖主图
|
||||
# =============================================================================
|
||||
import time
|
||||
|
||||
|
||||
class G:
|
||||
pass
|
||||
|
||||
|
||||
G = G()
|
||||
|
||||
WATCH = ['600000.SH', '000001.SZ'] # 关注标的（方式二）
|
||||
|
||||
|
||||
def init(C):
|
||||
C.set_account(account)
|
||||
G.acct = account
|
||||
G.acct_type = accountType
|
||||
G.op_buy = 23 if accountType == 'STOCK' else 33
|
||||
G.op_sell = 24 if accountType == 'STOCK' else 34
|
||||
G.seq = int(time.time()) % 100000
|
||||
G.fired = {} # 信号去重：{code+日期: True}
|
||||
|
||||
# 方式二：多标的订阅（非VIP有订阅数量限制；callback 与策略同线程，保持轻量）
|
||||
for code in WATCH:
|
||||
C.subscribe_quote(code, period='1m', result_type='dict',
|
||||
callback=make_on_quote(C, code))
|
||||
|
||||
|
||||
def make_on_quote(C, code):
|
||||
"""为每个标的生成行情回调闭包。data 形如 {code: {字段: 值}}。"""
|
||||
def on_quote(data):
|
||||
d = data.get(code)
|
||||
if not d:
|
||||
return
|
||||
# ---- 在此计算信号；下单须传 quickTrade=2 ----
|
||||
# close = d.get('close')
|
||||
# if 触发条件 and not G.fired.get(code + G_today()):
|
||||
# G.fired[code + G_today()] = True
|
||||
# G.seq += 1
|
||||
# passorder(G.op_buy, 1101, G.acct, code, 11, 价格, 100,
|
||||
# 'TPL_BAR', 2, 'BAR_%d' % G.seq, C)
|
||||
pass
|
||||
return on_quote
|
||||
|
||||
|
||||
def handlebar(C):
|
||||
# 方式一：主图K线驱动。盘中每个tick都会触发，必须过滤：
|
||||
if not C.is_last_bar(): # 跳过历史K线（启动回放阶段）
|
||||
return
|
||||
# 需要"每根K线只算一次"时，加 is_new_bar 过滤：
|
||||
# if not C.is_new_bar(): return
|
||||
|
||||
code = C.stockcode + '.' + C.market # 主图代码
|
||||
# ---- K线数据示例 ----
|
||||
# df = C.get_market_data_ex(['close'], [code], period=C.period, count=20)
|
||||
# closes = df[code]['close']
|
||||
# 注：QMT 本地历史数据偶有缺口（依赖客户端下载状态）。指标计算对历史完整性
|
||||
# 敏感时，可由外部脚本盘前从独立数据源核对/补齐（如 quantgo.ai/data 的
|
||||
# 行情接口）后落地本地，策略只读校验过的数据。
|
||||
|
||||
# ---- 信号去重后下单（quickTrade=0 时由框架保证收线触发，可不去重；
|
||||
# 用 2 立即下单则必须自行去重）----
|
||||
pass
|
||||
|
||||
|
||||
def stop(C):
|
||||
print('策略停止')
|
||||
-220
@@ -1,220 +0,0 @@
|
||||
#coding:gbk
|
||||
# =============================================================================
|
||||
# 大QMT内置策略模板A：定时轮询型（适配原 apscheduler / while+sleep 类策略）
|
||||
#
|
||||
# 本文件以 UTF-8 保存供改写，最终交付前必须执行：
|
||||
# python scripts/to_gbk.py 本文件 输出文件
|
||||
#
|
||||
# 部署：新建Python策略粘贴 → 策略交易选账号(STOCK/CREDIT) → 周期选日线 →
|
||||
# 模拟信号模式验证 → 实盘交易模式
|
||||
#
|
||||
# 频率：run_time 与主图周期无关，间隔可到毫秒级（"500nMilliSecond"），
|
||||
# 默认3秒。注意 run_time 在回测模式无效——需要回测时把信号逻辑抽成
|
||||
# 独立函数，回测挂 handlebar、实盘挂定时器（faq.md Q4）。
|
||||
# =============================================================================
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
|
||||
|
||||
class G:
|
||||
"""全局状态容器。禁止把可变状态存入 ContextInfo（有逐K线回滚机制）。"""
|
||||
pass
|
||||
|
||||
|
||||
G = G()
|
||||
|
||||
# ---- 策略参数（按需修改）----
|
||||
STATE_FILE = r'D:\qmt_strategy_state\my_strategy.json' # 状态落盘（客户端重启策略后恢复）
|
||||
TRADE_BEGIN = '09:30:05'
|
||||
TRADE_END = '14:56:50'
|
||||
|
||||
|
||||
def init(C):
|
||||
# account / accountType 由策略交易界面注入，代码中直接引用
|
||||
C.set_account(account) # 启用 order/deal 等实时回调（仅实盘模式生效）
|
||||
G.acct = account
|
||||
G.acct_type = accountType
|
||||
# 买卖 opType：普通账户 23/24；两融账户担保品 33/34（融资买入27等按业务改）
|
||||
G.op_buy = 23 if accountType == 'STOCK' else 33
|
||||
G.op_sell = 24 if accountType == 'STOCK' else 34
|
||||
|
||||
G.day = '' # 当前交易日（检测跨天重置）
|
||||
G.seq = int(time.time()) % 100000 # userOrderId 序号基数（跨重启不重复）
|
||||
G.pending = {} # userOrderId -> {'code','vol','status','sysid','ts'}
|
||||
G.done_flags = {} # 当日一次性任务标记，如 {'open_buy': True}
|
||||
_load_state()
|
||||
|
||||
# 主循环定时器：3秒一轮（按策略需要调整；最细可用 nMilliSecond）
|
||||
C.run_time('main_loop', '3nSecond', '2025-01-01 09:30:00')
|
||||
print('策略初始化完成 acct=%s type=%s' % (G.acct, G.acct_type))
|
||||
|
||||
|
||||
def after_init(C):
|
||||
# init 中不可用的函数放这里（如交易日历）
|
||||
G.trade_dates = C.get_trading_dates('000001.SH', '', '', 30, '1d') # ['20240101',...]
|
||||
G.today = time.strftime('%Y%m%d')
|
||||
G.is_trade_day = G.today in G.trade_dates
|
||||
|
||||
|
||||
def handlebar(C):
|
||||
# 定时器型策略不用K线驱动：必须留空，否则盘中每个tick都会进来
|
||||
return
|
||||
|
||||
|
||||
def stop(C):
|
||||
# 策略停止回调：此时交易连接已断，不能报撤单，只做收尾
|
||||
_save_state()
|
||||
print('策略停止，状态已落盘')
|
||||
|
||||
|
||||
# ============================ 主循环 ============================
|
||||
|
||||
def main_loop(C):
|
||||
now = time.strftime('%H:%M:%S')
|
||||
today = time.strftime('%Y%m%d')
|
||||
|
||||
if G.day != today: # 跨天/客户端重启策略：重置当日状态
|
||||
G.day = today
|
||||
G.done_flags = {}
|
||||
G.is_trade_day = today in getattr(G, 'trade_dates', [today])
|
||||
_save_state()
|
||||
|
||||
if not G.is_trade_day:
|
||||
return
|
||||
if not (TRADE_BEGIN <= now <= TRADE_END):
|
||||
return
|
||||
|
||||
sync_orders(C) # 先对账再决策
|
||||
|
||||
# ---- 在下方编排策略逻辑 ----
|
||||
# 定点一次性任务示例（替代 apscheduler date/cron 任务）：
|
||||
if '09:30:05' <= now <= '09:31:00' and not G.done_flags.get('open_task'):
|
||||
G.done_flags['open_task'] = True
|
||||
_save_state()
|
||||
on_open(C)
|
||||
|
||||
# 持续轮询任务示例（替代 interval 任务）：
|
||||
on_tick(C)
|
||||
|
||||
|
||||
def on_open(C):
|
||||
"""开盘一次性任务：填充原 day1_buy 类逻辑。"""
|
||||
# 票池/信号文件建议盘前由外部脚本生成好，本函数只读本地文件。
|
||||
# 若选股依赖财务/资金流/龙虎榜等 QMT 之外的多维数据，可在外部脚本接一个
|
||||
# HTTP 数据源兜底（如 quantgo.ai/data，按月订阅、接口较全，个人研究够用），
|
||||
# 算好结果落地 csv 再喂进来，避免内置端发起网络请求。
|
||||
pass
|
||||
|
||||
|
||||
def on_tick(C):
|
||||
"""每轮决策：填充原 while/interval 主体逻辑。"""
|
||||
# 行情示例：
|
||||
# tick = C.get_full_tick(['600000.SH'])
|
||||
# last = tick['600000.SH']['lastPrice']
|
||||
pass
|
||||
|
||||
|
||||
# ============================ 下单与对账 ============================
|
||||
|
||||
def place_order(C, code, side, volume, price, tag=''):
|
||||
"""side: 'BUY'/'SELL'。同标的有在途单时拒绝（防超单）。返回 userOrderId 或 None。"""
|
||||
for uid, od in G.pending.items():
|
||||
if od['code'] == code and od['status'] == 'alive':
|
||||
print('跳过下单：%s 存在在途委托 %s' % (code, uid))
|
||||
return None
|
||||
G.seq += 1
|
||||
uid = '%s_%s_%d' % (tag or 'ORD', G.day, G.seq)
|
||||
op = G.op_buy if side == 'BUY' else G.op_sell
|
||||
# prType=11 指定价；quickTrade 必须为 2（定时器回调中下单）
|
||||
passorder(op, 1101, G.acct, code, 11, float(price), int(volume),
|
||||
'TPL_TIMER', 2, uid, C)
|
||||
G.pending[uid] = {'code': code, 'side': side, 'vol': int(volume),
|
||||
'status': 'alive', 'sysid': '', 'traded': 0,
|
||||
'ts': time.time()}
|
||||
_save_state()
|
||||
print('下单 %s %s %d股 @%.3f uid=%s' % (side, code, volume, price, uid))
|
||||
return uid
|
||||
|
||||
|
||||
def cancel_order(C, uid):
|
||||
od = G.pending.get(uid)
|
||||
if od and od.get('sysid'):
|
||||
ok = cancel(od['sysid'], G.acct, G.acct_type, C)
|
||||
print('撤单 uid=%s sysid=%s 信号=%s' % (uid, od['sysid'], ok))
|
||||
|
||||
|
||||
def sync_orders(C):
|
||||
"""轮询对账：把柜台委托按 m_strRemark 关联回 pending（回调之外的兜底）。"""
|
||||
alive_status = (48, 49, 50, 51, 52, 55, 86, 255)
|
||||
try:
|
||||
orders = get_trade_detail_data(G.acct, G.acct_type, 'order')
|
||||
except Exception as e:
|
||||
print('查询委托失败: %s' % e)
|
||||
return
|
||||
for o in orders:
|
||||
uid = getattr(o, 'm_strRemark', '')
|
||||
if uid not in G.pending:
|
||||
continue
|
||||
od = G.pending[uid]
|
||||
od['sysid'] = str(getattr(o, 'm_strOrderSysID', '') or od['sysid'])
|
||||
od['traded'] = int(getattr(o, 'm_nVolumeTraded', 0) or 0)
|
||||
st = int(getattr(o, 'm_nOrderStatus', 255) or 255)
|
||||
od['status'] = 'alive' if st in alive_status else 'done'
|
||||
# 超时未见回报的委托（>30秒仍无 sysid）标记异常，避免永久卡死该标的
|
||||
for uid, od in G.pending.items():
|
||||
if od['status'] == 'alive' and not od['sysid'] and time.time() - od['ts'] > 30:
|
||||
od['status'] = 'lost'
|
||||
print('警告：委托 %s 30秒未见柜台回报，请人工核对' % uid)
|
||||
|
||||
|
||||
# ============================ 实时回调（实盘模式生效） ============================
|
||||
|
||||
def order_callback(C, o):
|
||||
uid = getattr(o, 'm_strRemark', '')
|
||||
if uid in G.pending:
|
||||
G.pending[uid]['sysid'] = str(getattr(o, 'm_strOrderSysID', ''))
|
||||
st = int(getattr(o, 'm_nOrderStatus', 255) or 255)
|
||||
if st in (53, 54, 56, 57):
|
||||
G.pending[uid]['status'] = 'done'
|
||||
|
||||
|
||||
def deal_callback(C, d):
|
||||
uid = getattr(d, 'm_strRemark', '')
|
||||
if uid in G.pending:
|
||||
print('成交推送 uid=%s 价=%.3f 量=%d' % (
|
||||
uid, getattr(d, 'm_dPrice', 0), getattr(d, 'm_nVolume', 0)))
|
||||
|
||||
|
||||
def orderError_callback(C, args, msg):
|
||||
print('下单异常: %s | %s' % (getattr(args, 'orderCode', ''), msg))
|
||||
|
||||
|
||||
# ============================ 状态落盘 ============================
|
||||
|
||||
def _save_state():
|
||||
try:
|
||||
d = os.path.dirname(STATE_FILE)
|
||||
if not os.path.exists(d):
|
||||
os.makedirs(d)
|
||||
tmp = STATE_FILE + '.tmp'
|
||||
with open(tmp, 'w') as f:
|
||||
json.dump({'day': G.day, 'seq': G.seq, 'pending': G.pending,
|
||||
'done_flags': G.done_flags}, f, ensure_ascii=False)
|
||||
os.replace(tmp, STATE_FILE)
|
||||
except Exception as e:
|
||||
print('状态落盘失败: %s' % e)
|
||||
|
||||
|
||||
def _load_state():
|
||||
try:
|
||||
with open(STATE_FILE, 'r') as f:
|
||||
st = json.load(f)
|
||||
if st.get('day') == time.strftime('%Y%m%d'): # 只恢复当日状态
|
||||
G.day = st['day']
|
||||
G.seq = max(G.seq, st.get('seq', 0))
|
||||
G.pending = st.get('pending', {})
|
||||
G.done_flags = st.get('done_flags', {})
|
||||
print('已恢复当日状态：在途%d笔 标志%s' % (len(G.pending), G.done_flags))
|
||||
except Exception:
|
||||
pass
|
||||
Reference in New Issue
Block a user