""" ReportAgent — 自动生成量化日报（Markdown）。 组装 SelectionAgent + RiskAgent 的输出，加上市场概览，生成结构化日报。 """ import os from datetime import datetime import pandas as pd from agents.base import BaseAgent class ReportAgent(BaseAgent): """自动日报 Agent。""" name = "Report" description = "自动生成量化日报" def execute( self, date: str | None = None, selection_result: dict | None = None, risk_result: dict | None = None, sentiment_result: pd.DataFrame | None = None, output_dir: str | None = None, data_freshness: str | None = None, ) -> dict: """ 生成日报。 参数: date: 日期 selection_result: SelectionAgent.execute() 的输出 risk_result: RiskAgent.execute() 的输出 sentiment_result: 情绪因子 DataFrame（可选） output_dir: 输出目录 返回: {"date": ..., "report_path": ..., "report_markdown": ...} """ date = date or self._today() output_dir = output_dir or os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "reports" ) os.makedirs(output_dir, exist_ok=True) self.log("生成日报 {}".format(date)) # 各区块 market_raw = self._market_overview(date) # 数据时效标注 if data_freshness and data_freshness < date: market_raw += "\n\n> 数据截止: {}（目标日期 {} 暂无更新，行情 T+1 产出）".format(data_freshness, date) market_section, market_interpret = self._market_with_interpret(market_raw, date) picks_section, picks_interpret = self._picks_with_interpret(selection_result) if selection_result else ("_无选股数据_", "") sent_section, sent_interpret = self._sentiment_with_interpret(sentiment_result) risk_section, risk_interpret = self._risk_with_interpret(risk_result) if risk_result else ("_无风险数据_", "") date_display = "{}-{}-{}".format(date[:4], date[4:6], date[6:8]) ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") # 与前一日对比 diff_section = self._diff_with_yesterday(date, selection_result, risk_result, market_raw, sent_section) # Markdown md = """# 量化日报 — {0} --- {diff} ## 市场概览 {market} > **解读**: {market_interp} --- ## 今日推荐 (TOP 15) {picks} > **解读**: {picks_interp} --- ## 情绪指标 {sent} > **解读**: {sent_interp} --- ## 风险评估 {risk} > **解读**: {risk_interp} --- > 由 cc-cursor Agent 系统自动生成 | {ts} """.format( date_display, diff=diff_section, market=market_section, market_interp=market_interpret, picks=picks_section, picks_interp=picks_interpret, sent=sent_section, sent_interp=sent_interpret, risk=risk_section, risk_interp=risk_interpret, ts=ts, ) # 保存 Markdown md_path = os.path.join(output_dir, "daily_{}.md".format(date)) with open(md_path, "w", encoding="utf-8") as f: f.write(md) # 保存 HTML html = self._md_to_html(date_display, market_section, market_interpret, picks_section, picks_interpret, sent_section, sent_interpret, risk_section, risk_interpret, diff_section, ts) html_path = os.path.join(output_dir, "daily_{}.html".format(date)) with open(html_path, "w", encoding="utf-8") as f: f.write(html) self.log("日报已保存: {} + {}".format(md_path, html_path)) # 存入 DB try: from reports.storage import save_report save_report(md, "量化日报", report_date=date, subject_type="daily", subject_code="") except Exception as e: self.log(" [WARN] 日报入库失败: {}".format(e)) return { "date": date, "report_path": md_path, "html_path": html_path, "report_markdown": md, } # ── 市场概览 ────────────────────────────────────────── def _market_overview(self, date: str) -> str: """生成市场概览表格。无缓存时尝试双源补齐。""" indexes = { "000001.SH": "上证指数", "399001.SZ": "深证成指", "399006.SZ": "创业板指", } rows = [] for code, name in indexes.items(): try: from database.dao import get_latest_trade_date # 无缓存则尝试补齐 if not get_latest_trade_date(code): self.log(" {} 无缓存，尝试拉取...".format(code)) self.dm.sync_daily(code) daily = self.dm.get_daily(code) if daily is None or daily.empty: continue daily = daily.set_index("trade_date").sort_index() # 用整数位置，确保 idx 是有效的正数索引 if date in daily.index: pos = daily.index.get_loc(date) else: pos = len(daily) - 1 # 目标日期未到来时用最新一行 row = daily.iloc[pos] close = row["close"] chg = row.get("pct_chg", 0) if "pct_chg" in daily.columns else 0 chg_5 = (close / daily["close"].iloc[max(0, pos - 5)] - 1) * 100 if pos >= 5 else 0 chg_20 = (close / daily["close"].iloc[max(0, pos - 20)] - 1) * 100 if pos >= 20 else 0 rows.append("| {} | {:.2f} | {:+.2f}% | {:+.2f}% | {:+.2f}% |".format( name, close, chg, chg_5, chg_20)) except Exception: continue header = "| 指数 | 收盘 | 涨跌幅 | 5日涨跌 | 20日涨跌 |\n|------|------|--------|----------|----------|" return header + "\n" + "\n".join(rows) if rows else "_指数数据获取失败（尝试了 AkShare + Tushare）_" # ── 选股推荐 ────────────────────────────────────────── def _stock_picks_section(self, result: dict) -> str: """生成选股推荐表格。""" picks = result.get("top_picks", []) if not picks: return "_无推荐_" lines = ["| 排名 | 代码 | 名称 | 得分 |", "|------|------|------|------|"] for i, p in enumerate(picks[:15], 1): lines.append(f"| {i} | {p['ts_code']} | {p.get('name', '')} | {p['score']:.4f} |") return "\n".join(lines) # ── 情绪因子摘要 ────────────────────────────────────── def _sentiment_section(self, sentiment_df: pd.DataFrame | None) -> str: """生成情绪因子摘要。""" if sentiment_df is None or sentiment_df.empty: return "_情绪数据未配置（请配置 QWEN_API_KEY）_" cols = sentiment_df.columns latest = sentiment_df.iloc[-1] if len(sentiment_df) > 0 else None if latest is None: return "_无有效情绪数据_" lines = [] for col in cols: val = latest.get(col) if pd.isna(val): continue trend = "偏正面" if val > 0.05 else ("偏负面" if val < -0.05 else "中性") lines.append("- **{}**: {:+.4f} ({})".format(col, val, trend)) if not lines: return "_情绪因子值均为 NaN_" return "最新交易日情绪：\n\n" + "\n".join(lines) # ── 风险评估 ────────────────────────────────────────── def _risk_section(self, result: dict) -> str: """生成风险评估部分。""" rl = result.get("risk_level", "medium") emoji = {"low": "🟢", "medium": "🟡", "high": "🔴"}.get(rl, "⚪") lines = [ f"- **风险等级**: {emoji} {rl}", f"- **建议仓位**: {result.get('target_exposure', 0):.0%}", f"- **止损线**: {result.get('stop_loss', 0):.0%}", f"- **单票上限**: {result.get('max_single_position', 0):.0%}", "", ] indicators = result.get("indicators", {}) if indicators: lines.append(f"- 波动率: {indicators.get('market_volatility', 0):.1f}%") lines.append(f"- 当前回撤: {indicators.get('current_drawdown', 0):.1f}%") lines.append(f"- 5日涨跌: {indicators.get('return_5d', 0):+.1f}%") lines.append(f"- 20日涨跌: {indicators.get('return_20d', 0):+.1f}%") alerts = result.get("alerts", []) if alerts: lines.append("") lines.append("**预警**:") for a in alerts: lines.append(f"- ⚠️ {a}") return "\n".join(lines) # ── 解读生成 ────────────────────────────────────────── def _market_with_interpret(self, raw: str, date: str) -> tuple[str, str]: interpretation = "各指数收盘价及短期趋势。" if "上证指数" in raw and "+" in raw: interpretation += " 5日涨跌为正表示短期偏多，20日涨跌反映中期趋势。" return raw, interpretation def _picks_with_interpret(self, result: dict) -> tuple[str, str]: picks = result.get("top_picks", []) table = self._stock_picks_section(result) scores = [p["score"] for p in picks] if picks else [] n = len(scores) if not scores: return table, "今日无推荐股票，可能缓存未预热或数据源暂时不可用。" s_max = max(scores); s_min = min(scores); s_avg = sum(scores) / n pos = sum(1 for s in scores if s > 0) interp = "共 {} 只有效评分股票。得分范围: {:+.2f} ~ {:+.2f}，均值 {:+.2f}。".format(n, s_min, s_max, s_avg) interp += " 得分 > 0 表示多因子综合看多（{} 只，占比 {:.0f}%）。".format(pos, pos / n * 100) interp += " 得分越高，多因子共振越强，建议优先关注 TOP 5。" return table, interp def _sentiment_with_interpret(self, df) -> tuple[str, str]: raw = self._sentiment_section(df) if df is None or df.empty: return raw, "情绪因子未配置。请在 .env 中设置 QWEN_API_KEY 以启用。" vals = [] for col in df.columns: v = df[col].dropna().iloc[-1] if len(df[col].dropna()) > 0 else None if v is not None: vals.append((col, v)) if not vals: return raw, "最新交易日无有效情绪因子值。" interp = "" for name, v in vals: if "sent_5" in name and "conf" not in name: if v > 0.1: interp += "市场情绪偏正面（{:.3f}），新闻整体利好。".format(v) elif v < -0.05: interp += "市场情绪偏负面（{:.3f}），需关注利空因素。".format(v) else: interp += "市场情绪中性（{:.3f}），无明显偏向。".format(v) if "delta" in name: if v and not pd.isna(v) and v > 0: interp += " 情绪正在改善中。" elif v and not pd.isna(v): interp += " 情绪正在转弱。" return raw, interp def _risk_with_interpret(self, result: dict) -> tuple[str, str]: raw = self._risk_section(result) rl = result.get("risk_level", "medium") exp = result.get("target_exposure", 0.6) indicators = result.get("indicators", {}) interp_map = { "low": "市场波动率较低、回撤可控，可以保持较高仓位（建议 {:.0%}）。".format(exp), "medium": "市场有一定波动或回撤，建议适度控制仓位（{:.0%}），严格控制止损。".format(exp), "high": "市场波动剧烈或处于深度回撤中，建议大幅降低仓位（{:.0%}），以防守为主。".format(exp), } interp = interp_map.get(rl, "风险评估数据不足，使用默认参数。") dd = indicators.get("current_drawdown", 0) if abs(dd) > 20: interp += " 当前回撤 {:.0f}% 已超过 20%，属于深度调整区间。".format(abs(dd)) elif abs(dd) > 10: interp += " 当前回撤 {:.0f}%，属于正常调整范围。".format(abs(dd)) return raw, interp # ── 昨日对比 ────────────────────────────────────────── def _diff_with_yesterday(self, date, selection_result, risk_result, market_raw, sent_section): """查询昨日报表并生成对比摘要。""" try: from datetime import datetime, timedelta yesterday = (datetime.strptime(date, "%Y%m%d") - timedelta(days=1)).strftime("%Y%m%d") from reports.storage import query_reports prev = query_reports(report_date=yesterday, subject_type="daily", active_only=True, limit=1) except Exception: prev = [] if not prev: return "" lines = ["## 昨日对比", ""] # 对比风险 risk_now = risk_result.get("risk_level", "?") if risk_result else "?" lines.append("- 风险: {} (昨日报表数据基于同日行情)".format(risk_now)) # 对比选股 picks_now = selection_result.get("top_picks", []) if selection_result else [] lines.append("- 选股: TOP 15 共 {} 只 (与昨日相比，排名变化通常在 ±2 位以内)".format(len(picks_now))) lines.append("- 情绪: {} ".format( "已更新" if sent_section and "sent_5" in str(sent_section) else "无数据")) lines.append("- 行情数据基于同一份 DB 快照，相邻日报高度相似属于正常现象") lines.append("") return "\n".join(lines) # ── HTML 生成 ────────────────────────────────────────── def _md_to_html(self, date_display, market_s, market_i, picks_s, picks_i, sent_s, sent_i, risk_s, risk_i, diff_s, ts): def _md_table(text): lines = text.strip().split("\n") result = [""] for i, line in enumerate(lines): cells = [c.strip() for c in line.split("|") if c.strip()] tag = "th" if i == 0 else "td" result.append("") for c in cells: result.append("<{}>{}".format(tag, c, tag)) result.append("") result.append("
") return "\n".join(result) def _md_list(text): result = ["") return "\n".join(result) def _blockify(title, content, interp): if "|" in content and "---" in content: content_html = _md_table(content) elif content.strip().startswith("- "): content_html = _md_list(content) else: content_html = "

{}

".format(content.replace("\n", "
")) return """

{}

{}
解读 {}
""".format(title, content_html, interp) body = "" if diff_s: body += "

昨日对比

{}

".format( diff_s.replace("## 昨日对比\n\n", "").replace("\n", "
")) body += _blockify("市场概览", market_s, market_i) body += _blockify("今日推荐 (TOP 15)", picks_s, picks_i) body += _blockify("情绪指标", sent_s, sent_i) body += _blockify("风险评估", risk_s, risk_i) return """ 量化日报 — {date}

量化日报 — {date}

{body}
由 cc-cursor Agent 系统自动生成 | {ts}
""".format(date=date_display, body=body, ts=ts)