433 lines
16 KiB
Python
433 lines
16 KiB
Python
import os
|
||
import re
|
||
import argparse
|
||
import unicodedata
|
||
from dataclasses import dataclass, field
|
||
from typing import List, Dict, Optional
|
||
from pathlib import Path
|
||
|
||
|
||
def sanitize_path_name(name: str) -> str:
|
||
name = unicodedata.normalize('NFKD', name)
|
||
name = name.encode('ascii', 'ignore').decode('ascii')
|
||
name = re.sub(r'[^\w\-_.]', '_', name)
|
||
name = re.sub(r'_{2,}', '_', name)
|
||
return name.strip('_')
|
||
|
||
|
||
@dataclass
|
||
class SetParameter:
|
||
name: str
|
||
param_type: str
|
||
default: float
|
||
min_value: float
|
||
max_value: float
|
||
step: float
|
||
optimize: bool
|
||
|
||
|
||
@dataclass
|
||
class SetFileResult:
|
||
file_path: str
|
||
ea_name: str
|
||
parameters: Dict[str, SetParameter] = field(default_factory=dict)
|
||
optimize_count: int = 0
|
||
fixed_count: int = 0
|
||
|
||
|
||
MT5_NATIVE_PATTERN = re.compile(r'^(\w+)\s+<([^>]+)>\s+<([^>]+)>')
|
||
OUR_FORMAT_PATTERN = re.compile(r'^(\w+)=([^|]+)\|\|([^|]+)\|\|([^|]+)\|\|([^|]+)\|\|([YN])')
|
||
|
||
|
||
def detect_encoding(file_path: str) -> str:
|
||
with open(file_path, 'rb') as f:
|
||
raw = f.read(4)
|
||
if raw[:2] == b'\xff\xfe':
|
||
return 'utf-16-le'
|
||
elif raw[:2] == b'\xfe\xff':
|
||
return 'utf-16-be'
|
||
elif raw[:3] == b'\xef\xbb\xbf':
|
||
return 'utf-8-sig'
|
||
return 'utf-8'
|
||
|
||
|
||
def detect_set_format(line: str) -> str:
|
||
if '=' in line and '||' in line:
|
||
return 'our_format'
|
||
elif '<' in line and '>' in line:
|
||
return 'mt5_native'
|
||
elif line.strip().startswith(';'):
|
||
return 'comment'
|
||
elif line.strip() == '':
|
||
return 'empty'
|
||
return 'unknown'
|
||
|
||
|
||
def parse_set_line(line: str) -> Optional[SetParameter]:
|
||
line = line.strip()
|
||
if not line or line.startswith(';'):
|
||
return None
|
||
|
||
fmt = detect_set_format(line)
|
||
if fmt == 'our_format':
|
||
m = OUR_FORMAT_PATTERN.match(line)
|
||
if m:
|
||
name, cur, start, step, stop, opt = m.groups()
|
||
|
||
cur_lower = cur.lower()
|
||
if cur_lower in ('true', 'false'):
|
||
is_true = cur_lower == 'true'
|
||
return SetParameter(
|
||
name=name,
|
||
param_type='bool',
|
||
default=1.0 if is_true else 0.0,
|
||
min_value=0.0,
|
||
max_value=1.0,
|
||
step=1.0,
|
||
optimize=(opt.upper() == 'Y')
|
||
)
|
||
|
||
return SetParameter(
|
||
name=name,
|
||
param_type='double',
|
||
default=float(cur),
|
||
min_value=float(start),
|
||
max_value=float(stop),
|
||
step=float(step),
|
||
optimize=(opt.upper() == 'Y')
|
||
)
|
||
elif fmt == 'mt5_native':
|
||
m = MT5_NATIVE_PATTERN.match(line)
|
||
if m:
|
||
name, val1, val2 = m.groups()
|
||
val1l = val1.lower().strip()
|
||
val2l = val2.lower().strip()
|
||
|
||
if val1l in ('true', 'false') or val2l in ('true', 'false'):
|
||
is_true = val1l == 'true' or val2l == 'true'
|
||
return SetParameter(
|
||
name=name,
|
||
param_type='bool',
|
||
default=1.0 if is_true else 0.0,
|
||
min_value=0.0,
|
||
max_value=1.0,
|
||
step=1.0,
|
||
optimize=False
|
||
)
|
||
|
||
try:
|
||
v1 = float(val1)
|
||
v2 = float(val2)
|
||
return SetParameter(
|
||
name=name,
|
||
param_type='double',
|
||
default=v1,
|
||
min_value=min(v1, v2),
|
||
max_value=max(v1, v2),
|
||
step=abs(v2 - v1) if v1 != v2 else 0,
|
||
optimize=False
|
||
)
|
||
except ValueError:
|
||
pass
|
||
return None
|
||
|
||
|
||
def parse_set_file(file_path: str) -> SetFileResult:
|
||
result = SetFileResult(file_path=file_path, ea_name='')
|
||
|
||
path = Path(file_path)
|
||
result.ea_name = path.stem.replace('_optimization', '').replace('_params', '')
|
||
|
||
enc = detect_encoding(file_path)
|
||
with open(file_path, 'r', encoding=enc) as f:
|
||
content = f.read()
|
||
|
||
for line in content.split('\n'):
|
||
param = parse_set_line(line)
|
||
if param:
|
||
result.parameters[param.name] = param
|
||
|
||
result.optimize_count = sum(1 for p in result.parameters.values() if p.optimize)
|
||
result.fixed_count = sum(1 for p in result.parameters.values() if not p.optimize)
|
||
|
||
return result
|
||
|
||
|
||
def parse_set_content(content: str) -> SetFileResult:
|
||
result = SetFileResult(file_path='memory', ea_name='')
|
||
|
||
for line in content.split('\n'):
|
||
param = parse_set_line(line)
|
||
if param:
|
||
result.parameters[param.name] = param
|
||
|
||
result.optimize_count = sum(1 for p in result.parameters.values() if p.optimize)
|
||
result.fixed_count = sum(1 for p in result.parameters.values() if not p.optimize)
|
||
|
||
return result
|
||
|
||
|
||
def scan_folder(folder_path: str) -> List[SetFileResult]:
|
||
results = []
|
||
folder = Path(folder_path)
|
||
|
||
if folder.is_file():
|
||
return [parse_set_file(str(folder))]
|
||
|
||
for set_file in folder.rglob('*.set'):
|
||
try:
|
||
result = parse_set_file(str(set_file))
|
||
results.append(result)
|
||
except Exception as e:
|
||
print(f'Warning: Failed to parse {set_file}: {e}')
|
||
|
||
return sorted(results, key=lambda x: x.ea_name)
|
||
|
||
|
||
def batch_scan(file_paths: List[str]) -> List[SetFileResult]:
|
||
results = []
|
||
for path in file_paths:
|
||
if os.path.exists(path):
|
||
try:
|
||
result = parse_set_file(path)
|
||
results.append(result)
|
||
except Exception as e:
|
||
print(f'Warning: Failed to parse {path}: {e}')
|
||
return results
|
||
|
||
|
||
def generate_prompt(results: List[SetFileResult], output_dir: str = '.') -> List[str]:
|
||
output_files = []
|
||
|
||
for result in results:
|
||
lines = []
|
||
lines.append('=' * 80)
|
||
lines.append(f'EA 参数优化分析请求 - {result.ea_name}')
|
||
lines.append('=' * 80)
|
||
lines.append('')
|
||
lines.append('## 基本信息')
|
||
lines.append(f'- EA名称: {result.ea_name}')
|
||
lines.append(f'- SET文件: {result.file_path}')
|
||
lines.append(f'- 参数总数: {len(result.parameters)} 个')
|
||
lines.append('')
|
||
|
||
lines.append('## 参数详情表')
|
||
lines.append('')
|
||
lines.append('| 参数名 | 类型 | 当前值 | 最小值 | 最大值 | 步进 | 建议范围 |')
|
||
lines.append('|--------|------|--------|--------|--------|------|----------|')
|
||
|
||
for name, param in sorted(result.parameters.items()):
|
||
suggest = f'{param.min_value}~{param.max_value}'
|
||
if param.param_type == 'int':
|
||
suggest = f'{int(param.min_value)}~{int(param.max_value)} (步进:{int(param.step)})'
|
||
elif param.param_type == 'double':
|
||
suggest = f'{param.min_value}~{param.max_value} (步进:{param.step})'
|
||
|
||
lines.append(f'| {name} | {param.param_type} | {param.default} | {param.min_value} | {param.max_value} | {param.step} | {suggest} |')
|
||
|
||
lines.append('')
|
||
lines.append('## 参数分析任务')
|
||
lines.append('')
|
||
lines.append('请分析以上所有参数，根据策略逻辑和交易逻辑判断：')
|
||
lines.append('')
|
||
lines.append('1. **哪些参数应该参与优化（optimize: true）**')
|
||
lines.append(' - 通常是影响策略核心逻辑、盈亏比、风险的关键参数')
|
||
lines.append(' - 例如：手数、止损止盈倍数、加仓间隔、风控阈值等')
|
||
lines.append('')
|
||
lines.append('2. **哪些参数应该保持固定（optimize: false）**')
|
||
lines.append(' - 通常是风控类、开关类、显示类参数')
|
||
lines.append(' - 例如：开关标识、魔数、面板位置等')
|
||
lines.append('')
|
||
lines.append('3. **推荐优化范围调整**')
|
||
lines.append(' - 检查当前范围是否合理')
|
||
lines.append(' - 给出你认为更合适的 min/max/step')
|
||
lines.append('')
|
||
lines.append('4. **重要参数优先级**')
|
||
lines.append(' - 哪些2-3个参数对策略影响最大，需要重点优化')
|
||
lines.append('')
|
||
lines.append('## 输出要求')
|
||
lines.append('')
|
||
lines.append('请输出符合以下格式的完整JSON配置文件：')
|
||
lines.append('')
|
||
lines.append('```json')
|
||
lines.append('{')
|
||
lines.append(f' "ea_name": "{result.ea_name}",')
|
||
safe_ea_name = sanitize_path_name(result.ea_name)
|
||
lines.append(f' "ea_path": "MY-EA\\\\{safe_ea_name}.ex5", // {result.ea_name}')
|
||
lines.append(' "description": "Edited via GUI",')
|
||
lines.append(' "search_strategy": "auto",')
|
||
lines.append(' "parameters": {')
|
||
param_list = list(result.parameters.items())
|
||
for i, (name, param) in enumerate(param_list):
|
||
comma = ',' if i < len(param_list) - 1 else ''
|
||
lines.append(f' "{name}": {{')
|
||
lines.append(f' "type": "{param.param_type}",')
|
||
lines.append(f' "default": {param.default},')
|
||
lines.append(f' "min": {param.min_value},')
|
||
lines.append(f' "max": {param.max_value},')
|
||
lines.append(f' "step": {param.step},')
|
||
lines.append(f' "optimize": true // 请根据分析填写 true 或 false')
|
||
lines.append(f' }}{comma}')
|
||
lines.append(' },')
|
||
lines.append(' "test_config": {')
|
||
lines.append(' "symbol": "XAUUSD", // 请填写交易品种')
|
||
lines.append(' "period": "M1", // 请填写周期: M1/M5/M15/H1/H4/D1')
|
||
lines.append(' "from_date": "2026.01.01", // 请填写开始日期')
|
||
lines.append(' "to_date": "2026.01.04", // 请填写结束日期')
|
||
lines.append(' "model": 0, // 建模方式: 0=Every Tick, 1=1分钟OHLC, 2=仅开盘价, 3=数学计算, 4=真实Tick')
|
||
lines.append(' "execution_delay": 0, // 执行延迟: 0=无延迟, -1=随机延迟, 正数=固定延迟毫秒')
|
||
lines.append(' "optimization_mode": 2, // 优化模式: 0=禁用, 1=慢速完整算法, 2=快速遗传算法, 3=MarketWatch所有符号')
|
||
lines.append(' "optimization_criterion": 1, // 优化指标: 0=余额最大, 1=盈利因子最大, 2=期望收益, 3=回撤最小, 4=恢复因子, 5=夏普比率, 6=自定义, 7=复合指标')
|
||
lines.append(' "deposit": 10000, // 初始保证金')
|
||
lines.append(' "leverage": "1:100", // 杠杆')
|
||
lines.append(' "timeout_min": 60 // 等待优化结果的超时(分钟)')
|
||
lines.append(' },')
|
||
lines.append(' "walk_forward": {')
|
||
lines.append(' "enabled": true, // 是否启用前向测试')
|
||
lines.append(' "forward_mode": 2, // 前向模式: 0=禁用, 1=OOS 50%, 2=OOS 33%, 3=OOS 25%, 4=自定义日期')
|
||
lines.append(' "forward_date": "" // 仅当 forward_mode=4 时必填, 格式 YYYY.MM.DD')
|
||
lines.append(' }')
|
||
lines.append('}')
|
||
lines.append('```')
|
||
lines.append('')
|
||
lines.append('注意：')
|
||
lines.append(f'- EA原始名称: {result.ea_name}')
|
||
lines.append(f'- ea_path 使用ASCII安全名称: {safe_ea_name}')
|
||
lines.append('- 实际EA文件应放在 MY-EA 目录下，文件名需与 ea_path 中的名称一致')
|
||
lines.append('- 每个参数的 "optimize" 字段需要你根据策略分析来填写')
|
||
lines.append('- optimize: true = 参与优化, optimize: false = 保持固定')
|
||
lines.append('- test_config 和 walk_forward 部分请根据回测需求填写或调整')
|
||
lines.append('- ea_path 路径使用双反斜杠 `\\\\`')
|
||
lines.append('')
|
||
lines.append('=' * 80)
|
||
|
||
output_path = os.path.join(output_dir, f'{result.ea_name}_optimization_prompt.txt')
|
||
with open(output_path, 'w', encoding='utf-8') as f:
|
||
f.write('\n'.join(lines))
|
||
|
||
output_files.append(output_path)
|
||
print(f'Generated: {output_path}')
|
||
|
||
return output_files
|
||
|
||
|
||
def interactive_scan():
|
||
print('请选择扫描模式：')
|
||
print('1. 扫描单个 SET 文件')
|
||
print('2. 扫描文件夹下所有 SET 文件')
|
||
print('3. 批量扫描多个 SET 文件')
|
||
print('4. 从剪贴板读取 SET 内容')
|
||
print('0. 退出')
|
||
print('')
|
||
|
||
choice = input('请输入选项 (0-4): ').strip()
|
||
|
||
if choice == '1':
|
||
path = input('请输入 SET 文件路径: ').strip()
|
||
if os.path.exists(path):
|
||
results = [parse_set_file(path)]
|
||
output_dir = os.path.dirname(path) or '.'
|
||
generate_prompt(results, output_dir)
|
||
else:
|
||
print('文件不存在')
|
||
|
||
elif choice == '2':
|
||
path = input('请输入文件夹路径: ').strip()
|
||
if os.path.exists(path):
|
||
results = scan_folder(path)
|
||
generate_prompt(results, path)
|
||
else:
|
||
print('文件夹不存在')
|
||
|
||
elif choice == '3':
|
||
print('请输入 SET 文件路径（每行一个，输入空行结束）:')
|
||
paths = []
|
||
while True:
|
||
line = input().strip()
|
||
if not line:
|
||
break
|
||
if os.path.exists(line):
|
||
paths.append(line)
|
||
else:
|
||
print(f'文件不存在: {line}')
|
||
if paths:
|
||
results = batch_scan(paths)
|
||
generate_prompt(results, os.path.dirname(paths[0]) if len(paths) == 1 else '.')
|
||
|
||
elif choice == '4':
|
||
print('请粘贴 SET 文件内容（输入空行结束）:')
|
||
lines = []
|
||
while True:
|
||
line = input()
|
||
if not line.strip():
|
||
break
|
||
lines.append(line)
|
||
if lines:
|
||
content = '\n'.join(lines)
|
||
result = parse_set_content(content)
|
||
result.ea_name = input('请输入 EA 名称: ').strip() or 'EA'
|
||
generate_prompt([result], '.')
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(
|
||
description='SET 文件扫描工具 - 生成 AI 优化提示词',
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog='''
|
||
示例:
|
||
python scan_set_prompt.py --file C:\\path\\to\\EA.set
|
||
python scan_set_prompt.py --folder C:\\path\\to\\set_files
|
||
python scan_set_prompt.py --batch a.set b.set c.set
|
||
python scan_set_prompt.py --interactive
|
||
'''
|
||
)
|
||
|
||
group = parser.add_mutually_exclusive_group(required=True)
|
||
group.add_argument('--file', '-f', metavar='PATH', help='扫描单个 SET 文件')
|
||
group.add_argument('--folder', '-d', metavar='PATH', help='扫描文件夹下所有 SET 文件')
|
||
group.add_argument('--batch', '-b', nargs='+', metavar='PATH', help='批量扫描多个 SET 文件')
|
||
group.add_argument('--interactive', '-i', action='store_true', help='交互式扫描')
|
||
|
||
parser.add_argument('--output', '-o', metavar='DIR', default='.', help='输出目录 (默认当前目录)')
|
||
|
||
args = parser.parse_args()
|
||
|
||
if args.interactive:
|
||
interactive_scan()
|
||
return
|
||
|
||
results = []
|
||
|
||
if args.file:
|
||
if not os.path.exists(args.file):
|
||
print(f'Error: 文件不存在 {args.file}')
|
||
return
|
||
results = [parse_set_file(args.file)]
|
||
output_dir = os.path.dirname(args.file) or args.output
|
||
|
||
elif args.folder:
|
||
if not os.path.exists(args.folder):
|
||
print(f'Error: 文件夹不存在 {args.folder}')
|
||
return
|
||
results = scan_folder(args.folder)
|
||
output_dir = args.folder
|
||
|
||
elif args.batch:
|
||
existing = [p for p in args.batch if os.path.exists(p)]
|
||
if not existing:
|
||
print('Error: 所有文件都不存在')
|
||
return
|
||
results = batch_scan(existing)
|
||
output_dir = args.output
|
||
|
||
if not results:
|
||
print('Warning: 没有找到任何 SET 文件')
|
||
return
|
||
|
||
output_files = generate_prompt(results, output_dir)
|
||
print(f'\n完成！共生成 {len(output_files)} 个提示词文件')
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|