|
|
# -*- coding: utf-8 -*-
|
|
|
import json
|
|
|
import httpx
|
|
|
from datetime import datetime, timedelta
|
|
|
|
|
|
|
|
|
class OdooClient:
|
|
|
"""Odoo JSON-RPC 客户端类"""
|
|
|
|
|
|
def __init__(self, url, db_name, username, password):
|
|
|
"""
|
|
|
初始化 Odoo 客户端并自动登录
|
|
|
|
|
|
Args:
|
|
|
url: Odoo 服务器地址
|
|
|
db_name: 数据库名称
|
|
|
username: 用户名
|
|
|
password: 密码
|
|
|
"""
|
|
|
self.url = url
|
|
|
self.db_name = db_name
|
|
|
self.username = username
|
|
|
self.password = password
|
|
|
self.client = None
|
|
|
self.uid = None
|
|
|
|
|
|
# 自动登录
|
|
|
self.login()
|
|
|
|
|
|
def login(self):
|
|
|
"""登录并获取 uid 和 client"""
|
|
|
try:
|
|
|
self.client = httpx.Client(timeout=30.0)
|
|
|
|
|
|
# Odoo 登录
|
|
|
payload = {
|
|
|
"jsonrpc": "2.0",
|
|
|
"method": "call",
|
|
|
"params": {
|
|
|
"service": "common",
|
|
|
"method": "login",
|
|
|
"args": [self.db_name, self.username, self.password]
|
|
|
},
|
|
|
"id": 1
|
|
|
}
|
|
|
|
|
|
response = self.client.post(f"{self.url}/jsonrpc", json=payload)
|
|
|
result = response.json()
|
|
|
|
|
|
# 检查是否有错误
|
|
|
if "error" in result:
|
|
|
raise Exception(f"登录失败: {result['error']}")
|
|
|
|
|
|
# Odoo 的登录响应中，result 直接就是 uid
|
|
|
self.uid = result.get("result")
|
|
|
|
|
|
if not self.uid:
|
|
|
raise Exception("登录失败：未获取到UID")
|
|
|
|
|
|
print(f"登录成功，UID: {self.uid}")
|
|
|
return True
|
|
|
|
|
|
except Exception as e:
|
|
|
print(f"登录失败: {e}")
|
|
|
if self.client:
|
|
|
self.client.close()
|
|
|
self.client = None
|
|
|
self.uid = None
|
|
|
raise
|
|
|
|
|
|
def logout(self):
|
|
|
"""退出登录并关闭连接"""
|
|
|
if self.client:
|
|
|
self.client.close()
|
|
|
self.client = None
|
|
|
self.uid = None
|
|
|
print("已退出")
|
|
|
|
|
|
def search_data(self, model, domain, fields=None, order=None, limit=None):
|
|
|
"""
|
|
|
通用搜索方法
|
|
|
|
|
|
Args:
|
|
|
model: 模型名称
|
|
|
domain: 搜索条件列表
|
|
|
fields: 需要返回的字段列表
|
|
|
order: 排序规则
|
|
|
limit: 返回记录数量限制
|
|
|
|
|
|
Returns:
|
|
|
查询结果列表，失败返回 None
|
|
|
"""
|
|
|
if not self.client or not self.uid:
|
|
|
raise Exception("未登录或连接已断开")
|
|
|
|
|
|
# 构建参数
|
|
|
args = [domain]
|
|
|
if fields:
|
|
|
args.append(fields)
|
|
|
|
|
|
kwargs = {}
|
|
|
if order:
|
|
|
kwargs['order'] = order
|
|
|
if limit:
|
|
|
kwargs['limit'] = limit
|
|
|
|
|
|
payload = {
|
|
|
"jsonrpc": "2.0",
|
|
|
"method": "call",
|
|
|
"params": {
|
|
|
"service": "object",
|
|
|
"method": "execute_kw",
|
|
|
"args": [
|
|
|
self.db_name,
|
|
|
self.uid,
|
|
|
self.password,
|
|
|
model,
|
|
|
"search_read",
|
|
|
args,
|
|
|
kwargs
|
|
|
]
|
|
|
},
|
|
|
"id": 2
|
|
|
}
|
|
|
|
|
|
try:
|
|
|
response = self.client.post(f"{self.url}/jsonrpc", json=payload)
|
|
|
result = response.json()
|
|
|
|
|
|
if "error" in result:
|
|
|
print(f"查询失败: {result['error']}")
|
|
|
return None
|
|
|
|
|
|
return result.get("result", [])
|
|
|
|
|
|
except Exception as e:
|
|
|
print(f"查询异常: {e}")
|
|
|
return None
|
|
|
|
|
|
def __enter__(self):
|
|
|
"""上下文管理器入口"""
|
|
|
return self
|
|
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
|
"""上下文管理器出口，自动登出"""
|
|
|
self.logout()
|
|
|
|
|
|
|
|
|
# 使用 OdooClient 类重构后的函数
|
|
|
def fetch_local_performance(odoo_client, model, domain, fields, limit):
|
|
|
"""获取本地表现数据"""
|
|
|
# 执行搜索
|
|
|
result = odoo_client.search_data(model=model, domain=domain, fields=fields, order="id desc", limit=limit)
|
|
|
|
|
|
if result:
|
|
|
return result
|
|
|
else:
|
|
|
print("未获取到数据")
|
|
|
exit(1)
|
|
|
|
|
|
|
|
|
def deconstruction_data(all_data):
|
|
|
"""解构并输出数据"""
|
|
|
for idx, data in enumerate(all_data):
|
|
|
data = json.loads(data.get('performance'))
|
|
|
print(f"\n{'=' * 60}")
|
|
|
print(f"策略 #{idx + 1}")
|
|
|
print(f"{'=' * 60}")
|
|
|
|
|
|
# 基础信息
|
|
|
alpha_id = data.get("id")
|
|
|
alpha_type = data.get("type")
|
|
|
author = data.get("author")
|
|
|
name = data.get("name")
|
|
|
|
|
|
print(f"ID: {alpha_id}")
|
|
|
print(f"类型: {alpha_type}")
|
|
|
print(f"作者: {author}")
|
|
|
print(f"名称: {name if name else '未命名'}")
|
|
|
|
|
|
# Settings 配置
|
|
|
settings = data.get("settings", {})
|
|
|
print(f"\n【配置信息】")
|
|
|
print(f" 品种类型: {settings.get('instrumentType')}")
|
|
|
print(f" 地区: {settings.get('region')}")
|
|
|
print(f" 股票池: {settings.get('universe')}")
|
|
|
print(f" 延迟: {settings.get('delay')}")
|
|
|
print(f" 衰减: {settings.get('decay')}")
|
|
|
print(f" 中性化: {settings.get('neutralization')}")
|
|
|
print(f" 截断: {settings.get('truncation')}")
|
|
|
print(f" 时间区间: {settings.get('startDate')} -> {settings.get('endDate')}")
|
|
|
|
|
|
# 因子代码
|
|
|
regular = data.get('regular', {})
|
|
|
code = regular.get('code')
|
|
|
description = regular.get('description')
|
|
|
|
|
|
print(f"\n【因子代码】")
|
|
|
print(f" Code: {code}")
|
|
|
if description:
|
|
|
print(f" Desc: {description}")
|
|
|
print(f" 操作符数量: {regular.get('operatorCount')}")
|
|
|
|
|
|
# 样本内（IS）回测结果 - 重点
|
|
|
in_sample = data.get('is', {})
|
|
|
|
|
|
if in_sample:
|
|
|
print(f"\n【样本内回测结果 (IS)】")
|
|
|
print(f" 📉 PnL: {in_sample.get('pnl', 0):,.0f}")
|
|
|
print(f" 📊 总收益率: {in_sample.get('returns', 0) * 100:.2f}%")
|
|
|
print(f" 🎯 夏普比率: {in_sample.get('sharpe', 0):.2f}")
|
|
|
print(f" 💪 Fitness: {in_sample.get('fitness', 0):.2f}")
|
|
|
print(f" 📉 最大回撤: {in_sample.get('drawdown', 0) * 100:.2f}%")
|
|
|
print(f" 🔄 换手率: {in_sample.get('turnover', 0) * 100:.2f}%")
|
|
|
print(f" 📈 多头持仓数: {in_sample.get('longCount', 0)}")
|
|
|
print(f" 📉 空头持仓数: {in_sample.get('shortCount', 0)}")
|
|
|
print(f" 💰 保证金: {in_sample.get('margin', 0):.6f}")
|
|
|
print(f" 📅 回测区间: {in_sample.get('startDate')} -> {in_sample.get('endDate')}")
|
|
|
|
|
|
# 检查项结果 - 排序后输出
|
|
|
checks = in_sample.get('checks', [])
|
|
|
if checks:
|
|
|
print(f"\n 【检查项】（已排序：PASS → WARNING → FAIL → PENDING）")
|
|
|
|
|
|
# 定义排序优先级
|
|
|
def sort_priority(check):
|
|
|
result = check.get('result')
|
|
|
priority_order = {
|
|
|
'PASS': 1,
|
|
|
'WARNING': 2,
|
|
|
'FAIL': 3,
|
|
|
'PENDING': 4
|
|
|
}
|
|
|
return priority_order.get(result, 5)
|
|
|
|
|
|
# 排序
|
|
|
sorted_checks = sorted(checks, key=sort_priority)
|
|
|
|
|
|
# 输出排序后的检查项
|
|
|
for check in sorted_checks:
|
|
|
name_check = check.get('name')
|
|
|
result = check.get('result')
|
|
|
limit = check.get('limit')
|
|
|
value = check.get('value')
|
|
|
|
|
|
# 根据结果用不同符号标记
|
|
|
if result == 'PASS':
|
|
|
symbol = '✅'
|
|
|
elif result == 'FAIL':
|
|
|
symbol = '❌'
|
|
|
elif result == 'WARNING':
|
|
|
symbol = '⚠️'
|
|
|
else:
|
|
|
symbol = '⏳'
|
|
|
|
|
|
if limit is not None and value is not None:
|
|
|
print(f" {symbol} {name_check}: {result} (要求: {limit}, 实际: {value})")
|
|
|
else:
|
|
|
print(f" {symbol} {name_check}: {result}")
|
|
|
else:
|
|
|
print(f"\n【样本内回测结果 (IS)】")
|
|
|
print(f" ⚠️ 无数据（因子可能在样本内阶段就失败了）")
|
|
|
|
|
|
# 样本外（OS）结果 - 通常为空表示策略无效
|
|
|
out_sample = data.get('os', {})
|
|
|
train = data.get('train', {})
|
|
|
test = data.get('test', {})
|
|
|
|
|
|
if out_sample:
|
|
|
print(f"\n【样本外回测结果 (OS)】")
|
|
|
print(f" 夏普比率: {out_sample.get('sharpe', 0):.2f}")
|
|
|
else:
|
|
|
print(f"\n【样本外回测结果 (OS)】⚠️ 为空 - 因子可能未通过样本内测试")
|
|
|
|
|
|
if train:
|
|
|
print(f"\n【训练集结果 (Train)】")
|
|
|
print(f" 夏普比率: {train.get('sharpe', 0):.2f}")
|
|
|
|
|
|
if test:
|
|
|
print(f"\n【测试集结果 (Test)】")
|
|
|
print(f" 夏普比率: {test.get('sharpe', 0):.2f}")
|
|
|
|
|
|
# 主题和竞赛信息
|
|
|
themes = data.get('themes', [])
|
|
|
if themes:
|
|
|
print(f"\n【关联主题】")
|
|
|
for theme in themes:
|
|
|
print(f" - {theme.get('name')} (乘数: {theme.get('multiplier')}x)")
|
|
|
|
|
|
competitions = data.get('competitions', [])
|
|
|
if competitions:
|
|
|
print(f"\n【关联竞赛】")
|
|
|
for comp in competitions:
|
|
|
print(f" - {comp.get('name')}")
|
|
|
|
|
|
# 状态信息
|
|
|
print(f"\n【状态】")
|
|
|
print(f" 阶段: {data.get('stage')}")
|
|
|
print(f" 状态: {data.get('status')}")
|
|
|
print(f" 是否收藏: {data.get('favorite')}")
|
|
|
print(f" 创建时间: {data.get('dateCreated')}")
|
|
|
|
|
|
# 如果有分类信息
|
|
|
classifications = data.get('classifications', [])
|
|
|
if classifications:
|
|
|
print(f"\n【分类】")
|
|
|
for cls in classifications:
|
|
|
print(f" - {cls.get('name')}")
|
|
|
|
|
|
|
|
|
# 使用示例
|
|
|
if __name__ == "__main__":
|
|
|
# ============================== Odoo 连接配置 ====================================
|
|
|
ODOO_URL = "http://192.168.31.41:32000"
|
|
|
DB_NAME = "quantify"
|
|
|
USERNAME = "rpc"
|
|
|
PASSWORD = "aaaAAA111"
|
|
|
|
|
|
# ============================== 搜索设置 ====================================
|
|
|
days = 7
|
|
|
now = datetime.now()
|
|
|
today_zero = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
|
days_ago_zero = today_zero - timedelta(days=days)
|
|
|
# 模型名称
|
|
|
model = "alpha.expression.line"
|
|
|
# 搜索条件
|
|
|
domain = [('performance', '!=', '{}'), ('write_date', '>=', days_ago_zero.strftime('%Y-%m-%d %H:%M:%S'))]
|
|
|
# 搜索字段
|
|
|
fields = ['id', 'performance']
|
|
|
# 搜索数量限制
|
|
|
limit = 1
|
|
|
|
|
|
try:
|
|
|
odoo = OdooClient(ODOO_URL, DB_NAME, USERNAME, PASSWORD)
|
|
|
|
|
|
all_data = fetch_local_performance(odoo, model, domain, fields, limit)
|
|
|
deconstruction_data(all_data)
|
|
|
|
|
|
|
|
|
except Exception as e:
|
|
|
print(f"程序执行失败: {e}")
|
|
|
|
|
|
with OdooClient(ODOO_URL, DB_NAME, USERNAME, PASSWORD) as odoo:
|
|
|
all_data = fetch_local_performance(odoo)
|
|
|
deconstruction_data(all_data)
|
|
|
|
