import json from datetime import datetime from pathlib import Path def generate_screen_report(results, market_ctx=None, llm_report=None, output_dir=None, mode="confirm"): if output_dir is None: output_dir = Path(__file__).parent / "reports" output_dir.mkdir(exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M") filename = f"screen_{timestamp}.html" filepath = Path(output_dir) / filename rows_html = "" for _, row in results.iterrows(): prob = row.get("buy_prob", 0.5) ml_raw = row.get("ml_score", 0) prob_pct = f"{prob*100:.1f}%" ml_raw_pct = f"{ml_raw*100:.1f}%" if prob >= 0.65: badge_class = "badge-danger" advice = "强烈推荐" stars = "★★★" elif prob >= 0.55: badge_class = "badge-warning" advice = "可以买入" stars = "★★☆" elif prob >= 0.45: badge_class = "badge-info" advice = "观望" stars = "★☆☆" else: badge_class = "badge-secondary" advice = "不建议" stars = "☆☆☆" chg = row.get("change_pct", 0) chg_class = "text-success" if chg >= 0 else "text-danger" chg_icon = "▲" if chg >= 0 else "▼" rows_html += f""" {int(row.get('rank', 0))} {row.get('code', '')} {row.get('name', '')}
{prob_pct} ML原始 {ml_raw_pct} {chg_icon} {abs(chg):.2f}% {stars} {advice} """ market_html = "" if market_ctx: advance = market_ctx.get("market_advance_count", 0) decline = market_ctx.get("market_decline_count", 0) total = advance + decline ratio = advance / total * 100 if total > 0 else 50 market_html = f"""
上涨 {advance}
下跌 {decline}
上涨比例 {ratio:.1f}%
市场均涨幅 {market_ctx.get('market_mean_return', 0):.2f}%
""" llm_html = "" if llm_report: llm_html = f"""

LLM分析报告

{_md_to_html(llm_report)}
""" now = datetime.now().strftime("%Y-%m-%d %H:%M") if mode == "pre": mode_label = "盘前规划版" title_note = "基于昨日数据，预测今日走向 | 结合9:25竞价观察，10:00确认买入" tips = """
  • 本版为盘前规划名单，预测的是今日走向，尚不可直接买入
  • 9:15-9:25 观察名单股票的竞价强弱（高开/低开/竞价量）
  • 10:00 运行确认版（实时数据）决定最终买入，次日卖出
    • """ else: mode_label = "盘中确认版" title_note = "实时数据，预测次日走向 | 从前5名选2-3只买入，次日卖出" tips = """
  • 从前5名中选择2-3只买入（概率>55%优先）
  • 止损：亏损 > 5% 立即卖出 | 止盈：盈利 > 5% 考虑卖出
  • 每只股票仓位不超过总资金的30%
    • """ html = f""" A股超短期选股（{mode_label}） - {now}

    A股超短期选股报告 · {mode_label}

    ML + LLM 智能量化选股系统
    {now}
    {title_note}
    {market_html}

    选股结果 Top {len(results)}

    {rows_html}
    排名 代码 名称 买入概率 涨跌幅 建议

    操作建议

      {tips}
    {llm_html}
    A股超短期选股系统 | ML: LightGBM + XGBoost | LLM: DeepSeek
    仅供参考，不构成投资建议。股市有风险，入市需谨慎。
    """ filepath.write_text(html, encoding="utf-8") return str(filepath) def generate_analyze_report(symbol, stock_name, buy_prob, tech_indicators, tech_analysis, advice, llm_analysis=None, output_dir=None): if output_dir is None: output_dir = Path(__file__).parent / "reports" output_dir.mkdir(exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M") filename = f"analyze_{symbol}_{timestamp}.html" filepath = Path(output_dir) / filename prob_pct = f"{buy_prob*100:.1f}%" if buy_prob >= 0.65: rating = "强烈推荐买入" stars = "★★★" rating_color = "#ff4d4f" elif buy_prob >= 0.55: rating = "可以考虑买入" stars = "★★☆" rating_color = "#faad14" elif buy_prob >= 0.45: rating = "建议观望" stars = "★☆☆" rating_color = "#1890ff" else: rating = "不建议买入" stars = "☆☆☆" rating_color = "#8c8c8c" tech_rows = "" indicators = [ ("RSI(14)", "rsi_14", "{:.1f}"), ("MACD差值", "macd_diff", "{:.4f}"), ("布林带位置", "bb_pct", "{:.2f}"), ("5日量比", "vol_ratio_5", "{:.2f}"), ("ATR波动率", "atr_ratio", "{:.4f}"), ("ADX趋势", "adx_14", "{:.1f}"), ("MFI资金流", "mfi_14", "{:.1f}"), ("OBV趋势", "obv_ret", "{:.4f}"), ] for label, key, fmt in indicators: val = tech_indicators.get(key, 0) status = tech_analysis.get(key, "") if val and status: tech_rows += f""" {label} {fmt.format(val)} {status} """ now = datetime.now().strftime("%Y-%m-%d %H:%M") html = f""" {symbol} {stock_name} 分析报告

    {symbol} {stock_name}

    个股分析报告 | {now}

    ML预测

    {prob_pct}
    买入概率
    {stars} {rating}

    当前行情

    最新价
    {tech_indicators.get('close', 0):.2f}
    涨跌幅
    {tech_indicators.get('change_pct', 0):.2f}%

    技术指标

    {tech_rows}
    指标数值解读

    操作建议

    综合建议
    {advice.get('summary', '')}
    买入条件
    {advice.get('buy_condition', '')}
    止损位
    {advice.get('stop_loss', '')}
    止盈位
    {advice.get('take_profit', '')}
    {"

    LLM深度分析

    " + _md_to_html(llm_analysis) + "
    " if llm_analysis else ""}
    A股超短期选股系统 | 仅供参考，不构成投资建议
    """ filepath.write_text(html, encoding="utf-8") return str(filepath) def _md_to_html(md_text): if not md_text: return "" lines = md_text.split("\n") html_lines = [] in_table = False for line in lines: stripped = line.strip() if stripped.startswith("|") and stripped.endswith("|"): if not in_table: html_lines.append("") in_table = True cells = [c.strip() for c in stripped.split("|")[1:-1]] if all(set(c) <= set("- :") for c in cells): continue tag = "th" if not in_table else "td" row = "".join(f"<{tag}>{c}" for c in cells) html_lines.append(f"{row}") else: if in_table: html_lines.append("
    ") in_table = False if stripped.startswith("### "): html_lines.append(f"

    {stripped[4:]}

    ") elif stripped.startswith("## "): html_lines.append(f"

    {stripped[3:]}

    ") elif stripped.startswith("# "): html_lines.append(f"

    {stripped[2:]}

    ") elif stripped.startswith("- "): html_lines.append(f"
  • {stripped[2:]}
    • ") elif stripped.startswith("**") and stripped.endswith("**"): html_lines.append(f"

    {stripped[2:-2]}

    ") elif stripped: html_lines.append(f"

    {stripped}

    ") if in_table: html_lines.append("") return "\n".join(html_lines)