MoFin 初始提交
完整数据采集+分析管道: - market_watch.py:90行业板块采集(同花顺/东方财富) - 市场精选推荐 cron:全市场分析+候选池+星级推荐 - price_monitor.py:持仓/自选高频价格监控 - refresh_mtf_cache.py:多周期K线缓存 - 策略评估/知识萃取管道 文档:docs/ 含完整需求+架构设计 注意:尚未配置 git remote,笑笑接手后自行配置
This commit is contained in:
@@ -0,0 +1,115 @@
|
||||
"""提示词管理数据模型"""
|
||||
|
||||
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)
|
||||
Reference in New Issue
Block a user