Files
MoFin/prompt_manager/models.py
T
知微 (MoFin) aa0f740381 MoFin 初始提交
完整数据采集+分析管道:
- market_watch.py:90行业板块采集(同花顺/东方财富)
- 市场精选推荐 cron:全市场分析+候选池+星级推荐
- price_monitor.py:持仓/自选高频价格监控
- refresh_mtf_cache.py:多周期K线缓存
- 策略评估/知识萃取管道

文档:docs/ 含完整需求+架构设计
注意:尚未配置 git remote,笑笑接手后自行配置
2026-06-20 12:04:21 +08:00

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)