57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
"""投研观察记录。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from app.db.database import get_db
|
|
from app.db import tables
|
|
|
|
|
|
def _safe_json(data: dict[str, Any] | None) -> str:
|
|
if not data:
|
|
return "{}"
|
|
try:
|
|
return json.dumps(data, ensure_ascii=False, default=str)
|
|
except Exception:
|
|
return "{}"
|
|
|
|
|
|
async def save_research_observations(observations: list[dict[str, Any]]) -> None:
|
|
"""批量保存本轮候选股投研观察。
|
|
|
|
记录失败不能影响筛选主流程。
|
|
"""
|
|
if not observations:
|
|
return
|
|
try:
|
|
values = []
|
|
now = datetime.now()
|
|
for item in observations:
|
|
values.append({
|
|
"scan_session": item.get("scan_session") or "manual",
|
|
"scan_mode": item.get("scan_mode") or "",
|
|
"ts_code": item.get("ts_code") or "",
|
|
"name": item.get("name") or item.get("ts_code") or "",
|
|
"theme_name": item.get("theme_name") or "",
|
|
"stock_role": item.get("stock_role") or "",
|
|
"action_plan": item.get("action_plan") or "观察",
|
|
"final_score": float(item.get("final_score") or 0),
|
|
"catalyst_score": float(item.get("catalyst_score") or 0),
|
|
"theme_money_score": float(item.get("theme_money_score") or 0),
|
|
"stock_money_score": float(item.get("stock_money_score") or 0),
|
|
"emotion_role_score": float(item.get("emotion_role_score") or 0),
|
|
"timing_score": float(item.get("timing_score") or 0),
|
|
"entry_signal_type": item.get("entry_signal_type") or "none",
|
|
"elimination_reason": item.get("elimination_reason") or "",
|
|
"detail_json": _safe_json(item.get("detail")),
|
|
"created_at": now,
|
|
})
|
|
async with get_db() as db:
|
|
await db.execute(tables.research_observations_table.insert(), values)
|
|
await db.commit()
|
|
except Exception:
|
|
pass
|