DockerSpace / services /line_message_builder.py
DennisChan0909's picture
feat: show degraded mode warning in LINE prediction messages
cbc429b
Raw
History Blame Contribute Delete
8.36 kB
"""
LINE Flex Message builder — extracted from routers/line_webhook.py.
Builds rich Flex Message bubbles for stock prediction results.
"""
import logging
from typing import Optional
logger = logging.getLogger(__name__)
# Flex Message imports — wrapped so the module still loads on older SDK builds
try:
from linebot.v3.messaging import (
FlexMessage,
FlexBubble,
FlexBubbleStyles,
FlexBlockStyle,
FlexBox,
FlexText,
FlexSeparator,
FlexButton,
MessageAction,
QuickReply,
QuickReplyItem,
TextMessage,
)
FLEX_AVAILABLE = True
except ImportError:
FLEX_AVAILABLE = False
def prob_bar(prob: float, width: int = 10) -> str:
"""Return a text-based probability bar, e.g. '█████░░░░░ 72%'."""
filled = round(prob * width)
return "█" * filled + "░" * (width - filled) + f" {prob:.0%}"
def build_flex_prediction(
name: str,
stock_no: str,
exchange: str,
price_str: str,
signal: str,
confidence: str,
buy_prob: float,
sell_prob: float,
change_pct: float,
today_change_pct: Optional[float] = None,
degraded: bool = False,
):
"""
Build a LINE Flex Message bubble for a stock prediction result.
Raises ImportError if FLEX_AVAILABLE is False — caller must handle.
"""
if not FLEX_AVAILABLE:
raise ImportError("Flex Message classes not available in this SDK version")
# Header colour driven by signal
header_colours = {"BUY": "#27ACB2", "SELL": "#FF6B35", "HOLD": "#888888"}
header_bg = header_colours.get(signal, "#888888")
signal_labels = {"BUY": "📈 買進", "SELL": "📉 賣出", "HOLD": "⏸ 觀望"}
conf_labels = {"HIGH": "高 ★★★", "MEDIUM": "中 ★★☆", "LOW": "低 ★☆☆"}
signal_label = signal_labels.get(signal, signal)
conf_label = conf_labels.get(confidence, confidence)
exch_str = f"({exchange})" if exchange else ""
bar_buy = prob_bar(buy_prob)
bar_sell = prob_bar(sell_prob)
change_str = f"{change_pct:+.2f}%"
# -- Header --
header = FlexBox(
layout="vertical",
background_color=header_bg,
padding_all="lg",
contents=[
FlexText(
text=f"{name} {stock_no}{exch_str}",
weight="bold",
color="#FFFFFF",
size="lg",
wrap=True,
),
FlexText(
text="台股預測",
color="#FFFFFFCC",
size="xs",
),
],
)
# -- Body --
today_change_row = []
if today_change_pct is not None:
tc_str = f"{today_change_pct:+.2f}%"
tc_color = "#27ACB2" if today_change_pct >= 0 else "#FF6B35"
today_change_row = [
FlexBox(
layout="horizontal",
contents=[
FlexText(text="今日漲跌", color="#666666", size="sm", flex=2),
FlexText(text=tc_str, weight="bold", size="sm", flex=3, align="end", color=tc_color),
],
),
]
body = FlexBox(
layout="vertical",
spacing="md",
padding_all="lg",
contents=[
FlexBox(
layout="horizontal",
contents=[
FlexText(text="現價", color="#666666", size="sm", flex=2),
FlexText(text=price_str, weight="bold", size="sm", flex=3, align="end"),
],
),
*today_change_row,
FlexSeparator(),
FlexBox(
layout="horizontal",
contents=[
FlexText(text="訊號", color="#666666", size="sm", flex=2),
FlexText(text=signal_label, weight="bold", size="sm", flex=3, align="end"),
],
),
FlexBox(
layout="horizontal",
contents=[
FlexText(text="信心度", color="#666666", size="sm", flex=2),
FlexText(text=conf_label, size="sm", flex=3, align="end"),
],
),
FlexSeparator(),
FlexBox(
layout="horizontal",
contents=[
FlexText(text="買進", color="#27ACB2", size="sm", flex=2),
FlexText(text=bar_buy, size="sm", flex=5, align="end"),
],
),
FlexBox(
layout="horizontal",
contents=[
FlexText(text="賣出", color="#FF6B35", size="sm", flex=2),
FlexText(text=bar_sell, size="sm", flex=5, align="end"),
],
),
FlexSeparator(),
FlexBox(
layout="horizontal",
contents=[
FlexText(text="預測5日漲跌", color="#666666", size="sm", flex=3),
FlexText(
text=change_str,
weight="bold",
size="sm",
flex=2,
align="end",
color="#27ACB2" if change_pct >= 0 else "#FF6B35",
),
],
),
FlexSeparator(),
*(
[FlexText(
text="⚠️ 資料較少,預測準確度可能較低",
color="#FF6B35",
size="xs",
wrap=True,
)]
if degraded else []
),
FlexText(
text="⚠️ 僅供參考,非投資建議",
color="#AAAAAA",
size="xxs",
wrap=True,
),
],
)
# -- Footer --
footer = FlexBox(
layout="vertical",
contents=[
FlexButton(
action=MessageAction(label="再查一支", text=stock_no),
style="link",
color="#27ACB2",
height="sm",
),
],
)
bubble = FlexBubble(
header=header,
body=body,
footer=footer,
styles=FlexBubbleStyles(
header=FlexBlockStyle(background_color=header_bg),
),
)
return FlexMessage(alt_text=f"{name} {stock_no} 預測結果", contents=bubble)
def build_quick_reply_menu(results: list[dict]):
"""Build a Quick Reply menu from search results (max 13 items)."""
if not FLEX_AVAILABLE:
raise ImportError("Quick Reply classes not available in this SDK version")
items = []
for r in results[:13]:
label = f"{r['stock_no']} {r['name']}"[:20] # LINE limit: 20 chars
items.append(QuickReplyItem(action=MessageAction(label=label, text=r['stock_no'])))
msg = TextMessage(text="找到以下股票,請選擇:")
msg.quick_reply = QuickReply(items=items)
return msg
def build_plain_text_prediction(
name: str,
stock_no: str,
exchange: str,
price_str: str,
signal: str,
confidence: str,
buy_prob: float,
sell_prob: float,
change_pct: float,
today_change_pct: Optional[float] = None,
degraded: bool = False,
) -> str:
"""Build a plain-text fallback prediction reply."""
signal_emoji = {"BUY": "📈 買進", "SELL": "📉 賣出", "HOLD": "⏸ 觀望"}
conf_label = {"HIGH": "高 ★★★", "MEDIUM": "中 ★★☆", "LOW": "低 ★☆☆"}
exch_str = f"({exchange})" if exchange else ""
today_line = f"今日漲跌:{today_change_pct:+.2f}%\n" if today_change_pct is not None else ""
return (
f"📊 {name} {stock_no}{exch_str}\n"
f"現價:{price_str}\n"
f"{today_line}"
f"─────────────\n"
f"訊號:{signal_emoji.get(signal, signal)}\n"
f"信心度:{conf_label.get(confidence, confidence)}\n"
f"買進機率:{buy_prob:.0%} 賣出:{sell_prob:.0%}\n"
f"預測5日漲跌:{change_pct:+.2f}%\n"
f"─────────────\n"
+ ("⚠️ 資料較少,預測準確度可能較低\n" if degraded else "")
+ f"⚠️ 僅供參考,非投資建議"
)