"""服务端生成日K+成交量 PNG,供大模型视觉解读。""" import io from datetime import datetime from .kline_store import get_daily_candles async def render_daily_chart_png_async(symbol: str, limit: int = 300) -> bytes: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt import matplotlib.dates as mdates candles, _ = await get_daily_candles(symbol, limit) if not candles: raise ValueError(f"no klines for {symbol}") times = [datetime.fromtimestamp(c["time"] / 1000) for c in candles] opens = [c["open"] for c in candles] highs = [c["high"] for c in candles] lows = [c["low"] for c in candles] closes = [c["close"] for c in candles] vols = [c.get("quote_volume") or c.get("volume") or 0 for c in candles] fig, (ax1, ax2) = plt.subplots( 2, 1, figsize=(12, 7), gridspec_kw={"height_ratios": [3, 1]}, facecolor="#0d1118" ) fig.subplots_adjust(hspace=0.08) for i in range(len(candles)): t = times[i] o, h, l, cl = opens[i], highs[i], lows[i], closes[i] color = "#0ecb81" if cl >= o else "#f6465d" ax1.plot([t, t], [l, h], color=color, linewidth=0.8) ax1.add_patch( plt.Rectangle( (mdates.date2num(t) - 0.3, min(o, cl)), 0.6, abs(cl - o) or 0.001, facecolor=color, edgecolor=color, ) ) ax1.set_facecolor("#0d1118") ax1.tick_params(colors="#8b9cb3") ax1.set_title(f"{symbol} 日K + 成交量", color="#e7ecf3", fontsize=14) ax1.grid(True, alpha=0.2) colors_vol = ["#0ecb81" if closes[i] >= opens[i] else "#f6465d" for i in range(len(candles))] ax2.bar(times, vols, color=colors_vol, alpha=0.7, width=0.8) ax2.set_facecolor("#0d1118") ax2.tick_params(colors="#8b9cb3") ax2.set_ylabel("成交额", color="#8b9cb3") ax2.grid(True, alpha=0.2) for ax in (ax1, ax2): ax.xaxis.set_major_formatter(mdates.DateFormatter("%m-%d")) fig.autofmt_xdate() buf = io.BytesIO() fig.savefig(buf, format="png", dpi=120, facecolor="#0d1118") plt.close(fig) buf.seek(0) return buf.read()