migration: preserve review journal alerts and assistant slice
This commit is contained in:
leefer
@@ -0,0 +1,188 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date, datetime, timedelta
|
||||
from typing import Any
|
||||
|
||||
from backend.bootstrap.config import normalize_date, tushare_code, validate_text
|
||||
from backend.data.providers.tushare_client import TushareError
|
||||
from backend.features.review.agent import ReviewAssistantError, stream_review_assistant
|
||||
|
||||
|
||||
class ReviewServiceMixin:
|
||||
def trade_entries(
|
||||
self, start_date: str = "", end_date: str = "", code: str = ""
|
||||
) -> dict[str, Any]:
|
||||
return self.trade_journal.list_entries(
|
||||
self.current_user_id, start_date, end_date, code
|
||||
)
|
||||
|
||||
def review_watchlist(self, trade_date: str) -> dict[str, Any]:
|
||||
normalized_date = normalize_date(trade_date)
|
||||
items = self.database.list_watchlist(self.current_user_id)
|
||||
if not items:
|
||||
return {"items": [], "trade_date": normalized_date}
|
||||
|
||||
resolved_date = normalized_date
|
||||
if self.configured:
|
||||
try:
|
||||
client = self._tushare_client()
|
||||
resolved_date, _ = client.resolve_trade_context(normalized_date)
|
||||
history = self.database.watchlist_price_history(
|
||||
[str(item["code"]) for item in items], resolved_date
|
||||
)
|
||||
missing_codes = [
|
||||
str(item["code"]) for item in items
|
||||
if len(history.get(str(item["code"])) or []) < 6
|
||||
]
|
||||
start_date = (
|
||||
datetime.strptime(resolved_date, "%Y%m%d") - timedelta(days=24)
|
||||
).strftime("%Y%m%d")
|
||||
for code in missing_codes:
|
||||
rows = client.query(
|
||||
"daily",
|
||||
{
|
||||
"ts_code": tushare_code(code),
|
||||
"start_date": start_date,
|
||||
"end_date": resolved_date,
|
||||
},
|
||||
"ts_code,trade_date,open,high,low,close,pct_chg,vol,amount",
|
||||
)
|
||||
if rows:
|
||||
self.database.upsert_daily_bars(rows)
|
||||
if missing_codes:
|
||||
history = self.database.watchlist_price_history(
|
||||
[str(item["code"]) for item in items], resolved_date
|
||||
)
|
||||
except (TushareError, ValueError):
|
||||
history = self.database.watchlist_price_history(
|
||||
[str(item["code"]) for item in items], resolved_date
|
||||
)
|
||||
else:
|
||||
history = self.database.watchlist_price_history(
|
||||
[str(item["code"]) for item in items], resolved_date
|
||||
)
|
||||
|
||||
auction_scores: dict[str, Any] = {}
|
||||
try:
|
||||
auction = self.auction_center(normalized_date, False)
|
||||
auction_scores = {
|
||||
str(row.get("code") or ""): row.get("attention_score")
|
||||
for row in (auction.get("watchlist_rows") or [])
|
||||
if row.get("available", True)
|
||||
}
|
||||
except (TushareError, ValueError):
|
||||
pass
|
||||
|
||||
enriched = []
|
||||
for item in items:
|
||||
code = str(item.get("code") or "")
|
||||
bars = history.get(code) or []
|
||||
latest = bars[-1] if bars else {}
|
||||
close = float(latest.get("close") or 0)
|
||||
base_close = float(bars[-6].get("close") or 0) if len(bars) >= 6 else 0
|
||||
enriched.append(
|
||||
{
|
||||
**item,
|
||||
"change": (
|
||||
round(float(latest.get("pct_chg") or 0), 2) if latest else None
|
||||
),
|
||||
"return_5d": (
|
||||
round((close / base_close - 1) * 100, 2)
|
||||
if close > 0 and base_close > 0 else None
|
||||
),
|
||||
"attention_score": auction_scores.get(code),
|
||||
"market_date": str(latest.get("trade_date") or ""),
|
||||
}
|
||||
)
|
||||
return {"items": enriched, "trade_date": resolved_date}
|
||||
|
||||
def save_trade_entry(self, payload: dict[str, Any]) -> dict[str, Any]:
|
||||
trade_id = self.trade_journal.save(self.current_user_id, payload)
|
||||
return {"id": trade_id, **self.trade_entries()}
|
||||
|
||||
def delete_trade_entry(self, trade_id: int) -> dict[str, Any]:
|
||||
deleted = self.trade_journal.delete(self.current_user_id, trade_id)
|
||||
return {"deleted": deleted, **self.trade_entries()}
|
||||
|
||||
def assistant_messages(self) -> list[dict[str, Any]]:
|
||||
return self.database.list_assistant_messages(self.current_user_id)
|
||||
|
||||
def clear_assistant_messages(self) -> int:
|
||||
return self.database.delete_assistant_messages(self.current_user_id)
|
||||
|
||||
def assistant_stream(self, payload: dict[str, Any]):
|
||||
question = validate_text(payload.get("question"), "问题", 2000, required=True)
|
||||
trade_date = normalize_date(
|
||||
str(payload.get("trade_date") or date.today().isoformat())
|
||||
)
|
||||
context = self._assistant_context(trade_date)
|
||||
history = [
|
||||
{"role": item["role"], "content": str(item["content"])[:4000]}
|
||||
for item in self.assistant_messages()[-12:]
|
||||
if item.get("role") in {"user", "assistant"}
|
||||
]
|
||||
def generate():
|
||||
answer_parts: list[str] = []
|
||||
events = self.llm_gateway.stream(
|
||||
"assistant",
|
||||
"review-assistant-v1",
|
||||
lambda profile: stream_review_assistant(
|
||||
context,
|
||||
question,
|
||||
history,
|
||||
profile.api_key,
|
||||
profile.base_url,
|
||||
profile.model,
|
||||
),
|
||||
(ReviewAssistantError,),
|
||||
)
|
||||
for event in events:
|
||||
if event.kind == "delta":
|
||||
chunk = str(event.value or "")
|
||||
answer_parts.append(chunk)
|
||||
yield chunk
|
||||
elif event.kind == "complete":
|
||||
self.database.save_assistant_exchange(
|
||||
self.current_user_id,
|
||||
question,
|
||||
"".join(answer_parts).strip(),
|
||||
trade_date,
|
||||
)
|
||||
|
||||
return generate()
|
||||
|
||||
def _assistant_context(self, trade_date: str) -> dict[str, Any]:
|
||||
dashboard = self.get_dashboard(trade_date)
|
||||
actual_date = normalize_date(
|
||||
str((dashboard.get("meta") or {}).get("trade_date") or trade_date)
|
||||
)
|
||||
sentiment = self.sentiment_history(actual_date, 10)
|
||||
tracking = self.strategy_tracking.list_tracking(self.current_user_id, 5)
|
||||
alerts = self.alert_service.list_alerts(
|
||||
self.current_user_id, "all", date.today().isoformat()
|
||||
)
|
||||
trades = self.trade_journal.list_entries(
|
||||
self.current_user_id, end_date=actual_date
|
||||
)
|
||||
return {
|
||||
"data_date": actual_date,
|
||||
"market": {
|
||||
"overview": dashboard.get("overview") or {},
|
||||
"top_sectors": (dashboard.get("sectors") or [])[:8],
|
||||
"limit_performance": dashboard.get("limit_performance") or {},
|
||||
"sentiment_history": (sentiment.get("rows") or [])[-10:],
|
||||
},
|
||||
"personal": {
|
||||
"watchlist": self.database.list_watchlist(self.current_user_id)[:30],
|
||||
"review_notes": self.database.list_notes(
|
||||
self.current_user_id, scope="daily"
|
||||
)[:10],
|
||||
"strategy_tracking": {
|
||||
"summary": tracking.get("summary") or {},
|
||||
"batches": (tracking.get("batches") or [])[:5],
|
||||
},
|
||||
"alerts": (alerts.get("items") or [])[:20],
|
||||
"trade_summary": trades.get("summary") or {},
|
||||
"trade_entries": (trades.get("items") or [])[:30],
|
||||
},
|
||||
}
|
||||
Reference in New Issue
Block a user