128 lines
4.4 KiB
Python
128 lines
4.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Stock story generation for explain view."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta
|
|
from typing import Any
|
|
|
|
from backend.data.market_store import MarketStore
|
|
|
|
|
|
def build_stock_story(
|
|
*,
|
|
symbol: str,
|
|
as_of_date: str,
|
|
price_rows: list[dict[str, Any]],
|
|
news_rows: list[dict[str, Any]],
|
|
) -> str:
|
|
"""Build a compact markdown story from enriched news and recent price action."""
|
|
lines = [f"## {symbol} Story", f"As of `{as_of_date}`"]
|
|
if not price_rows:
|
|
lines.append("")
|
|
lines.append("No OHLC data available for story generation.")
|
|
return "\n".join(lines)
|
|
|
|
open_price = float(price_rows[0].get("open") or price_rows[0].get("close") or 0.0)
|
|
close_price = float(price_rows[-1].get("close") or 0.0)
|
|
price_change = ((close_price - open_price) / open_price) * 100 if open_price else 0.0
|
|
high_price = max(float(row.get("high") or row.get("close") or 0.0) for row in price_rows)
|
|
low_price = min(float(row.get("low") or row.get("close") or 0.0) for row in price_rows)
|
|
|
|
lines.append("")
|
|
lines.append(
|
|
f"The stock moved {'up' if price_change >= 0 else 'down'} "
|
|
f"{abs(price_change):.2f}% over the recent window, trading between "
|
|
f"${low_price:.2f} and ${high_price:.2f}."
|
|
)
|
|
|
|
positive = [row for row in news_rows if str(row.get("sentiment") or "").lower() == "positive"]
|
|
negative = [row for row in news_rows if str(row.get("sentiment") or "").lower() == "negative"]
|
|
lines.append("")
|
|
lines.append(
|
|
f"Recent coverage included {len(news_rows)} relevant articles "
|
|
f"({len(positive)} positive / {len(negative)} negative)."
|
|
)
|
|
|
|
if news_rows:
|
|
lines.append("")
|
|
lines.append("### Key Moments")
|
|
ranked_rows = sorted(
|
|
news_rows,
|
|
key=lambda row: (
|
|
0 if str(row.get("relevance") or "").lower() in {"high", "relevant"} else 1,
|
|
-abs(float(row.get("ret_t0") or 0.0)),
|
|
),
|
|
)
|
|
for row in ranked_rows[:5]:
|
|
trade_date = row.get("trade_date") or str(row.get("date") or "")[:10]
|
|
title = row.get("title") or "Untitled"
|
|
key_discussion = row.get("key_discussion") or row.get("summary") or ""
|
|
sentiment = str(row.get("sentiment") or "neutral").lower()
|
|
lines.append(
|
|
f"- `{trade_date}` [{sentiment}] {title}: {str(key_discussion).strip()[:220]}"
|
|
)
|
|
|
|
if positive:
|
|
lines.append("")
|
|
lines.append("### Bullish Threads")
|
|
for row in positive[:3]:
|
|
reason = row.get("reason_growth") or row.get("key_discussion") or row.get("summary") or row.get("title")
|
|
lines.append(f"- {str(reason).strip()[:220]}")
|
|
|
|
if negative:
|
|
lines.append("")
|
|
lines.append("### Bearish Threads")
|
|
for row in negative[:3]:
|
|
reason = row.get("reason_decrease") or row.get("key_discussion") or row.get("summary") or row.get("title")
|
|
lines.append(f"- {str(reason).strip()[:220]}")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def get_or_create_stock_story(
|
|
store: MarketStore,
|
|
*,
|
|
symbol: str,
|
|
as_of_date: str,
|
|
) -> dict[str, Any]:
|
|
"""Return cached story or build a new one from recent market context."""
|
|
cached = store.get_story_cache(symbol, as_of_date=as_of_date)
|
|
if cached:
|
|
return {
|
|
"symbol": symbol,
|
|
"as_of_date": as_of_date,
|
|
"story": cached.get("content") or "",
|
|
"source": cached.get("source") or "cache",
|
|
}
|
|
|
|
start_date = None
|
|
if len(as_of_date) >= 10:
|
|
target_date = datetime.strptime(as_of_date[:10], "%Y-%m-%d").date()
|
|
start_date = (target_date - timedelta(days=29)).isoformat()
|
|
|
|
price_rows = (
|
|
store.get_ohlc(symbol, start_date, as_of_date)
|
|
if start_date
|
|
else []
|
|
)
|
|
news_rows = store.get_news_items_enriched(
|
|
symbol,
|
|
start_date=start_date,
|
|
end_date=as_of_date,
|
|
limit=40,
|
|
)
|
|
story = build_stock_story(
|
|
symbol=symbol,
|
|
as_of_date=as_of_date,
|
|
price_rows=price_rows,
|
|
news_rows=news_rows,
|
|
)
|
|
store.upsert_story_cache(symbol, as_of_date=as_of_date, content=story, source="local")
|
|
return {
|
|
"symbol": symbol,
|
|
"as_of_date": as_of_date,
|
|
"story": story,
|
|
"source": "local",
|
|
}
|