346 lines
11 KiB
Python
346 lines
11 KiB
Python
# -*- 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)
|