# -*- coding: utf-8 -*- """ 每日行情报告引擎（盘前 / 盘后） - premarket : 工作日 9:00 —— 昨日市场回顾 / 昨日至今要闻 / 全球市场 / 持仓与关注目标 / 盘前研判 - postmarket : 交易日 15:30 —— 今日市场总结 / 今日要闻 / 全球市场 / 持仓表现 / 盘后研判 每期输出两份报告： 简单版 —— 邮件正文（HTML，快速浏览） 详细版 —— HTML 附件（完整结构化 + AI 深度解读） """ import datetime as dt import html as html_mod import json import logging import time from database import query, query_one, execute from settings import mail_config, quiet_config, in_quiet_period from engine.analyst import llm_chat log = logging.getLogger("report") GLOBAL_ORDER = ["dji", "nasdaq", "sp500", "hsi", "nikkei", "kospi", "dax", "cac", "ftse"] KIND_META = { "premarket": {"name": "盘前分析", "scope": "昨日与今日", "title": "盘前 · 昨日市场回顾与今日展望"}, "postmarket": {"name": "盘后总结", "scope": "今日", "title": "盘后 · 今日市场总结"}, } # ===================================================================== 数据采集 def latest_trading_day(): r = query_one("SELECT MAX(date) d FROM stock_daily") return r["d"] if r else dt.date.today().isoformat() def collect_market(day): """指数 / 涨跌 / 量能 / 行业 / 个股""" idx = query("SELECT * FROM market_index WHERE date<=? ORDER BY date DESC LIMIT 2", (day,)) latest = idx[0] if idx else {} prev = idx[1] if len(idx) > 1 else latest inds = [] for k, label in (("sh", "上证指数"), ("sz", "深证成指"), ("cy", "创业板指")): cur, old = latest.get(k, 0), prev.get(k, 0) or 1 inds.append({"key": k, "label": label, "value": cur, "chg": round((cur - old) / old * 100, 2)}) stat = query_one( "SELECT COUNT(*) total, SUM(CASE WHEN change_pct>0 THEN 1 ELSE 0 END) up," "SUM(CASE WHEN change_pct<0 THEN 1 ELSE 0 END) down," "SUM(CASE WHEN change_pct>=9.8 THEN 1 ELSE 0 END) limit_up," "SUM(CASE WHEN change_pct<=-9.8 THEN 1 ELSE 0 END) limit_down," "ROUND(SUM(amount)/10000,2) amount_yi " "FROM stock_daily WHERE date=?", (day,)) heat = query( "SELECT s.industry, ROUND(AVG(d.change_pct),2) chg, COUNT(*) cnt " "FROM stock_daily d JOIN stocks s ON s.code=d.code WHERE d.date=? " "GROUP BY s.industry ORDER BY chg DESC", (day,)) gainers = query( "SELECT s.name, s.code, s.industry, d.change_pct FROM stock_daily d " "JOIN stocks s ON s.code=d.code WHERE d.date=? ORDER BY d.change_pct DESC LIMIT 8", (day,)) losers = query( "SELECT s.name, s.code, s.industry, d.change_pct FROM stock_daily d " "JOIN stocks s ON s.code=d.code WHERE d.date=? ORDER BY d.change_pct ASC LIMIT 8", (day,)) return {"date": day, "indexes": inds, "stat": stat, "heat": heat, "gainers": gainers, "losers": losers} def collect_news(since_date, limit=20): rows = query( "SELECT id,title,content,source,category,publish_date,sentiment,related_stocks FROM news " "WHERE publish_date>=? ORDER BY publish_date DESC, id DESC LIMIT ?", (since_date, limit)) # 按重要度排序（类别权重 + 情感强度） w = {"公司": 3, "业绩": 3, "机构观点": 2, "行业": 2, "市场": 1} for n in rows: n["_score"] = w.get(n["category"], 1) * 10 + abs(n["sentiment"]) * 5 rows.sort(key=lambda x: x["_score"], reverse=True) return rows def collect_positions(): rows = query( "SELECT w.code, s.name, s.industry, s.market_cap, d.close, d.change_pct " "FROM watchlist w JOIN stocks s ON s.code=w.code " "LEFT JOIN stock_daily d ON d.code=s.code AND d.date=(SELECT MAX(date) FROM stock_daily) " "ORDER BY w.added_at") out = [] for r in rows: sc = query_one( "SELECT AVG(sentiment) s FROM news WHERE (related_stocks=? OR related_stocks LIKE ? OR related_stocks LIKE ?) " "AND publish_date>=date('now','-7 day')", (r["code"], f"%,{r['code']}", f"{r['code']},%")) out.append({**r, "news_score": round(sc["s"], 2) if sc and sc["s"] is not None else 0}) return out def collect_targets(): tgts = query("SELECT id, type, code, name, keywords FROM watch_targets WHERE enabled=1") out = [] for t in tgts: if t["type"] == "stock" and t["code"]: latest = query_one( "SELECT meta, created_at FROM tracking_reports WHERE code=? ORDER BY id DESC LIMIT 1", (t["code"],)) else: latest = query_one( "SELECT meta, created_at FROM tracking_reports WHERE code=? ORDER BY id DESC LIMIT 1", (f"CONCEPT:{t['name']}",)) m = json.loads(latest["meta"]) if latest else {} out.append({"type": t["type"], "name": t["name"], "impact": m.get("impact_score"), "change_kind": m.get("change_kind"), "summary": m.get("summary", ""), "tracked_at": latest["created_at"] if latest else None}) return out def collect_global(): r = query_one("SELECT date, data FROM global_markets ORDER BY date DESC LIMIT 1") if not r: return [] try: data = json.loads(r["data"]) except Exception: return [] items = [] for k in GLOBAL_ORDER: if k in data: items.append(data[k]) return items # ===================================================================== 文本渲染 def fmt_market(mkt): s = mkt["stat"] or {} idx_txt = " ".join(f"{i['label']} {i['value']:.2f} ({i['chg']:+.2f}%)" for i in mkt["indexes"]) heat_txt = "、".join(f"{h['industry']}({h['chg']:+.2f}%)" for h in mkt["heat"][:6]) or "无" g_txt = "、".join(f"{g['name']}({g['change_pct']:+.2f}%)" for g in mkt["gainers"][:5]) l_txt = "、".join(f"{g['name']}({g['change_pct']:+.2f}%)" for g in mkt["losers"][:5]) return { "idx": idx_txt, "breadth": (f"上涨 {s.get('up',0)} / 下跌 {s.get('down',0)} 家，" f"涨停 {s.get('limit_up',0)} / 跌停 {s.get('limit_down',0)}，" f"两市成交 {s.get('amount_yi',0)} 亿"), "heat": heat_txt, "gainers": g_txt or "无", "losers": l_txt or "无", } def fmt_news(news, top=8): lines = [] for n in news[:top]: tone = "利好" if n["sentiment"] > 0 else ("利空" if n["sentiment"] < 0 else "中性") lines.append(f"- [{n['publish_date']}] {n['title']}（{n['category']}·{tone}{n['sentiment']:+.2f}）{n['content'][:60]}") return "\n".join(lines) or "（暂无）" def fmt_positions(pos): if not pos: return "（当前无持仓/自选股）" return "\n".join( f"- {p['name']}({p['code']}) {p['industry']} 收盘{p['close']} ({p['change_pct']:+.2f}%) 市值{p['market_cap']:.0f}亿 近7日消息面{p['news_score']:+.2f}" for p in pos) def fmt_targets(tgts): if not tgts: return "（当前无跟踪目标）" return "\n".join( f"- [{t['type']}] {t['name']} 影响度{t['impact'] or '--'}/100 {t['change_kind'] or ''} {t['summary'][:50]}" for t in tgts) def fmt_global(items): return " ".join(f"{g.get('label','')} {g.get('value',0):.2f} ({g.get('chg',0):+.2f}%)" for g in items) or "（暂无）" # ===================================================================== 生成报告 def _build_context(kind): day = latest_trading_day() mkt = collect_market(day) fm = fmt_market(mkt) if kind == "premarket": news = collect_news(day, limit=24) scope_txt = "昨日/最近交易日" else: news = collect_news(day, limit=24) scope_txt = "今日" pos = collect_positions() tgts = collect_targets() glob = collect_global() ctx = { "kind_name": KIND_META[kind]["name"], "date": day, "scope": scope_txt, "mkt": mkt, "fm": fm, "news": news, "news_txt": fmt_news(news, 10), "pos": pos, "pos_txt": fmt_positions(pos), "tgts": tgts, "tgts_txt": fmt_targets(tgts), "global_txt": fmt_global(glob), "global": glob, } return ctx def _base_prompt(ctx, detailed): d = ctx["date"] title = KIND_META[ctx["kind_name"] if ctx["kind_name"] in KIND_META else "premarket"]["title"] if False else "" kind = "盘前分析" if "盘前" in ctx["kind_name"] else "盘后总结" return f"""你是资深A股市场分析师，请基于下方【数据】生成一份{kind}报告。 【报告日期】{d} 【指数】{ctx['fm']['idx']} 【涨跌结构】{ctx['fm']['breadth']} 【领涨行业】{ctx['fm']['heat']} 【领涨个股】{ctx['fm']['gainers']} 【领跌个股】{ctx['fm']['losers']} 【重点要闻】 {ctx['news_txt']} 【全球市场】 {ctx['global_txt']} 【持仓/自选股】 {ctx['pos_txt']} 【关注目标/主题】 {ctx['tgts_txt']} """ def _brief_prompt(ctx): return _base_prompt(ctx, False) + """ 【输出要求】输出一份精炼的盘前/盘后速览（约 200-300 字），Markdown 格式，包含： 1. 一句话大盘研判 2. 3-5 条关键要点（行情/消息/持仓/主题） 3. 今日关注提示 要求信息密集、数据准确，不要编造数据。""" def _detail_prompt(ctx): return _base_prompt(ctx, True) + """ 【输出要求】输出一份完整的盘前/盘后分析报告（Markdown），结构如下： ## 一、市场概览（指数表现/涨跌结构/量能/领涨领跌板块个股解读） ## 二、消息面解析（分类解读重点要闻及影响：政策/行业/公司/机构观点） ## 三、全球市场联动（外围市场表现及对A股的传导） ## 四、持仓表现（逐只点评：涨跌、评分依据、近期消息面） ## 五、关注目标/主题（各主题/个股的最新动态与影响度解读） ## 六、操作策略与风险提示 数据须严格来自上文【数据】，可补充合理分析逻辑，不得编造数字。""" def generate_reports(kind): """生成 (brief_html, detail_html)""" ctx = _build_context(kind) brief_md = "" detail_md = "" try: brief_md = llm_chat([ {"role": "system", "content": "你是一名严谨专业的A股市场分析师。"}, {"role": "user", "content": _brief_prompt(ctx)}, ]).strip() except Exception as e: log.warning("brief llm fail: %s", e) try: detail_md = llm_chat([ {"role": "system", "content": "你是一名严谨专业的A股市场分析师。"}, {"role": "user", "content": _detail_prompt(ctx)}, ]).strip() except Exception as e: log.warning("detail llm fail: %s", e) brief_html = _render_brief(ctx, brief_md) detail_html = _render_detail(ctx, detail_md) return brief_html, detail_html # ===================================================================== 渲染 def _render_brief(ctx, brief_md): kind = KIND_META[ctx["kind_name"] if ctx["kind_name"] in KIND_META else "premarket"] rows = [] for i in ctx["mkt"]["indexes"]: rows.append(f"=0 else '#17a34a'}'>{i['label']} {i['value']:.2f} ({i['chg']:+.2f}%)") news_li = "".join(f"
  • [{n['publish_date']}] {html_mod.escape(n['title'])} ({n['category']})
    • " for n in ctx["news"][:6]) or "
  • 暂无
    • " pos_li = "".join(f"
  • {p['name']}({p['code']}) 收{p['close']} " f"=0 else '#17a34a'}'>{p['change_pct']:+.2f}% · {p['industry']}
    • " for p in ctx["pos"]) or "
  • 暂无持仓
    • " tgt_li = "".join(f"
  • {html_mod.escape(t['name'])}（影响度{t['impact'] or '--'}，{t['change_kind'] or '—'}）
    • " for t in ctx["tgts"][:5]) or "
  • 暂无目标
    • " gb = " ｜ ".join(f"{g.get('label','')} {g.get('value',0):.2f} " f"=0 else '#17a34a'}'>({g.get('chg',0):+.2f}%)" for g in ctx["global"][:6]) ai = html_mod.escape(brief_md) if brief_md else "（AI 简评生成失败，请查看附件详细版）" return f"""
    📊 智能荐股 · {kind['name']}
    {ctx['date']} · 简版速览 · 详细版见附件
    🔎 大盘：
    {'　'.join(rows)}
    {ctx['fm']['breadth']}
    领涨行业：{ctx['fm']['heat']}
    📰 重点要闻：
    🌏 全球市场：
    {gb}
    💼 持仓：
    🎯 关注目标/主题：
    🤖 AI 研判：
    {ai}
    智能荐股系统自动生成 · 内容基于模拟数据，仅供演示，不构成投资建议
    """ def _md_to_html(md): """极简 Markdown → HTML（用于附件详细版）""" md = html_mod.escape(md or "") out, in_list = [], False for line in md.splitlines(): line = line.rstrip() if not line: if in_list: out.append(""); in_list = False continue if line.startswith("## "): if in_list: out.append(""); in_list = False out.append(f"

    {line[3:]}

    ") elif line.startswith("### "): if in_list: out.append(""); in_list = False out.append(f"

    {line[4:]}

    ") elif line.startswith("##"): if in_list: out.append(""); in_list = False out.append(f"

    {line[2:].strip()}

    ") elif line.startswith("- "): if not in_list: out.append(""); in_list = False out.append(f"

    {line[2:]}

    ") else: if in_list: out.append(""); in_list = False out.append(f"

    {line}

    ") if in_list: out.append("") return "".join(out) def _render_detail(ctx, detail_md): kind = KIND_META[ctx["kind_name"] if ctx["kind_name"] in KIND_META else "premarket"] idx_rows = "".join( f"{i['label']}{i['value']:.2f}" f"=0 else '#17a34a'}'>{i['chg']:+.2f}%" for i in ctx["mkt"]["indexes"]) stat = ctx["mkt"]["stat"] or {} heat_rows = "".join(f"{h['industry']}{h['cnt']}" f"=0 else '#17a34a'}'>{h['chg']:+.2f}%" for h in ctx["mkt"]["heat"]) g_rows = "".join(f"{g['name']}{g['code']}" f"=0 else '#17a34a'}'>{g['change_pct']:+.2f}%" for g in ctx["mkt"]["gainers"]) l_rows = "".join(f"{g['name']}{g['code']}" f"=0 else '#17a34a'}'>{g['change_pct']:+.2f}%" for g in ctx["mkt"]["losers"]) news_rows = "".join( f"{n['publish_date']}{n['category']}{html_mod.escape(n['title'])}" f"=0 else '#17a34a'}'>{n['sentiment']:+.2f}" for n in ctx["news"][:15]) pos_rows = "".join( f"{p['name']}{p['code']}{p['industry']}{p['close']}" f"=0 else '#17a34a'}'>{p['change_pct']:+.2f}%" f"{p['market_cap']:.0f}亿{p['news_score']:+.2f}" for p in ctx["pos"]) or "暂无持仓" gb_rows = "".join(f"{g.get('label','')}{g.get('value',0):.2f}" f"=0 else '#17a34a'}'>{g.get('chg',0):+.2f}%" for g in ctx["global"]) body = _md_to_html(detail_md) if detail_md else "

    （AI 分析生成失败）

    " return f""" 智能荐股 · {kind['name']} {ctx['date']}

    📊 智能荐股 · {kind['name']}（{ctx['date']}）

    市场/要闻/全球/持仓/主题 全景分析 · 详细版报告 · 自动生成

    〇、数据总览

    指数{idx_rows}
    指数收盘涨跌
    涨跌结构：{ctx['fm']['breadth']}
    行业热度{heat_rows}
    行业家数平均涨跌
    领涨个股{g_rows}
    名称代码涨跌
    领跌个股{l_rows}
    名称代码涨跌

    重点要闻

    {news_rows}
    日期分类标题情感

    全球市场

    {gb_rows}
    指数点位涨跌

    持仓 / 自选股

    {pos_rows}
    股票行业收盘涨跌市值消息面

    关注目标 / 主题

    {html_mod.escape(ctx['tgts_txt']).replace(chr(10), '
    ')}

    AI 深度分析

    {body}
    智能荐股系统自动生成 · 内容基于模拟数据，仅供演示，不构成投资建议
    """ # ===================================================================== 发送 def send_daily_report(kind="premarket"): """生成并发送报告：正文=简版，附件=详细版 HTML。静默期内抑制发送。返回 dict 状态""" from engine.notifier import send_email # 静默期判断：命中则不生成、不发送，仅记一条 quiet 日志 qc = quiet_config("report") if qc["enabled"] and in_quiet_period(qc["ranges"]): meta = KIND_META.get(kind, KIND_META["premarket"]) subject = f"[智能荐股] {meta['name']} {time.strftime('%Y-%m-%d')}" execute("INSERT INTO report_log(kind, subject, brief_len, detail_len, status, message) " "VALUES(?,?,0,0,'quiet',?)", (kind, subject, "静默期内已抑制发送")) return {"ok": False, "quiet": True, "msg": "静默期内已抑制发送"} mc = mail_config() brief_html, detail_html = generate_reports(kind) meta = KIND_META.get(kind, KIND_META["premarket"]) subject = f"[智能荐股] {meta['name']} {time.strftime('%Y-%m-%d')}" detail_file = f"智能荐股_{meta['name']}_{time.strftime('%Y%m%d')}.html" try: send_email(subject, brief_html, cfg=mc, attachments=[{"filename": detail_file, "content": detail_html.encode("utf-8")}]) execute("INSERT INTO report_log(kind, subject, brief_len, detail_len, status, message) " "VALUES(?,?,?,?,'sent','附件: '||?)", (kind, subject, len(brief_html), len(detail_html), detail_file)) return {"ok": True, "subject": subject, "detail_file": detail_file} except Exception as e: execute("INSERT INTO report_log(kind, subject, brief_len, detail_len, status, message) " "VALUES(?,?,?,?,'failed',?)", (kind, subject, len(brief_html), len(detail_html), str(e))) return {"ok": False, "error": str(e)} def report_log(limit=20): return query("SELECT * FROM report_log ORDER BY id DESC LIMIT ?", (limit,)) if __name__ == "__main__": import sys kind = sys.argv[1] if len(sys.argv) > 1 else "premarket" if kind not in ("premarket", "postmarket"): kind = "premarket" logging.basicConfig(level=logging.INFO) r = send_daily_report(kind) print(r)