aa0f740381
完整数据采集+分析管道: - market_watch.py:90行业板块采集(同花顺/东方财富) - 市场精选推荐 cron:全市场分析+候选池+星级推荐 - price_monitor.py:持仓/自选高频价格监控 - refresh_mtf_cache.py:多周期K线缓存 - 策略评估/知识萃取管道 文档:docs/ 含完整需求+架构设计 注意:尚未配置 git remote,笑笑接手后自行配置
116 lines
4.1 KiB
Python
116 lines
4.1 KiB
Python
"""提示词管理数据模型"""
|
|
|
|
from dataclasses import dataclass, field, asdict
|
|
from typing import Optional
|
|
from datetime import datetime
|
|
import json
|
|
|
|
|
|
# ═══════════════════════════════════════════
|
|
# 枚举/常量
|
|
# ═══════════════════════════════════════════
|
|
|
|
PROMPT_CATEGORIES = {
|
|
"strategy": "策略生成 — 制定买入区/止损/止盈的规则集",
|
|
"scan": "快速盯盘 — 交易时段行情监控报告",
|
|
"evaluation": "策略评估 — 每日/每周策略效果评估",
|
|
"knowledge": "知识萃取 — 经验沉淀到知识日志",
|
|
"review": "持仓复查 — 定期全面持仓分析",
|
|
"health": "系统健康检查 — 每日开盘前检查",
|
|
"format": "报告格式规范 — 三段式输出格式规则",
|
|
"analysis": "分析规则 — 单股/行业分析流程",
|
|
}
|
|
|
|
VERSION_STATUS = {
|
|
"active": "当前使用中",
|
|
"deprecated": "已弃用(不再使用)",
|
|
"archived": "已归档(历史版本,仅保留记录)",
|
|
"experimental": "实验版本(临时测试)",
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════
|
|
# 数据类
|
|
# ═══════════════════════════════════════════
|
|
|
|
@dataclass
|
|
class PromptVersion:
|
|
"""单个提示词版本"""
|
|
version: str # v1, v2, v2.1 ...
|
|
label: str # 人类可读的名称
|
|
created_at: str # ISO datetime
|
|
changelog: str # 变更说明
|
|
content: str # 提示词完整内容(或摘要)
|
|
content_path: str = "" # 提示词文件路径(完整内容存储处)
|
|
author: str = "知微" # 修改者
|
|
status: str = "active" # active/deprecated/archived/experimental
|
|
tags: list = field(default_factory=list) # 标签:如 "技术面", "R/R阈值"
|
|
|
|
def to_dict(self):
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, d):
|
|
return cls(**d)
|
|
|
|
|
|
@dataclass
|
|
class PromptDef:
|
|
"""一条提示词定义"""
|
|
id: str # 唯一标识,如 "strategy-generation"
|
|
name: str # 显示名称
|
|
description: str # 用途说明
|
|
category: str # 分类
|
|
locations: list # 存放位置(文件路径、cron job ID等)
|
|
versions: list # PromptVersion 列表
|
|
created_at: str # 首次创建时间
|
|
updated_at: str # 最近更新时间
|
|
current_version: str # 当前活跃版本号
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"id": self.id,
|
|
"name": self.name,
|
|
"description": self.description,
|
|
"category": self.category,
|
|
"locations": self.locations,
|
|
"versions": [v.to_dict() for v in self.versions],
|
|
"created_at": self.created_at,
|
|
"updated_at": self.updated_at,
|
|
"current_version": self.current_version,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, d):
|
|
versions = [PromptVersion.from_dict(v) for v in d.get("versions", [])]
|
|
return cls(
|
|
id=d["id"],
|
|
name=d["name"],
|
|
description=d.get("description", ""),
|
|
category=d.get("category", ""),
|
|
locations=d.get("locations", []),
|
|
versions=versions,
|
|
created_at=d.get("created_at", ""),
|
|
updated_at=d.get("updated_at", ""),
|
|
current_version=d.get("current_version", ""),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class StrategyLink:
|
|
"""策略→提示词版本关联记录"""
|
|
code: str # 股票代码
|
|
name: str # 股票名称
|
|
prompt_id: str # 提示词ID
|
|
prompt_version: str # 提示词版本号
|
|
strategy_action: str # 生成的策略文本
|
|
generated_at: str # 生成时间
|
|
evaluation_result: Optional[dict] = None # 评估结果(后续补充)
|
|
|
|
def to_dict(self):
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, d):
|
|
return cls(**d)
|