diff --git a/backend/app/analysis/market_temp.py b/backend/app/analysis/market_temp.py index 45229a26..cc0a54b4 100644 --- a/backend/app/analysis/market_temp.py +++ b/backend/app/analysis/market_temp.py @@ -177,6 +177,10 @@ def calculate_market_temperature(trade_date: str = None) -> MarketTemperature: broken_rate=round(broken_rate, 1), index_above_ma20=index_above_ma20, temperature=round(temperature, 1), + source="tushare_daily", + data_status="fresh", + source_detail=f"daily={trade_date}", + limit_counts_reliable=not limit_df.empty, ) logger.info(f"市场温度 {trade_date}: {temperature:.1f} (涨{up_count}/跌{down_count}, 涨停{limit_up_count}, 连板{max_streak})") return result @@ -196,18 +200,25 @@ async def build_realtime_market_temperature( pct = sh_index.get("pct_chg", 0) if sh_index else 0 ratio = breadth.up_count / max(breadth.down_count, 1) - # 实时口径里，把涨停/跌停降级为可选增强项。 - # 主判断只依赖上涨/下跌家数与指数方向，避免涨停池接口不稳影响系统决策。 + # 实时口径里，涨跌停池可靠时用专门池；池接口失败时用全市场行情阈值估算。 + # 估算口径权重略低，并在研究报告中标记为降级，避免悄悄污染判断。 temp_from_ratio = min(ratio / 3.0 * 40, 40) - temp_from_limit_up = min(breadth.limit_up_count / 3, 10) if breadth.limit_counts_reliable else 0 + temp_from_limit_up = ( + min(breadth.limit_up_count / 3, 10) + if breadth.limit_counts_reliable + else min(breadth.limit_up_count / 5, 8) + ) temp_from_index = min(max(pct * 10 + 15, 0), 30) baseline.trade_date = breadth.trade_date baseline.up_count = breadth.up_count baseline.down_count = breadth.down_count - if breadth.limit_counts_reliable: - baseline.limit_up_count = breadth.limit_up_count - baseline.limit_down_count = breadth.limit_down_count + baseline.limit_up_count = breadth.limit_up_count + baseline.limit_down_count = breadth.limit_down_count baseline.temperature = round(min(max(temp_from_ratio + temp_from_limit_up + temp_from_index + 20, 0), 100), 1) baseline.index_above_ma20 = pct > 0 if sh_index else baseline.index_above_ma20 + baseline.source = breadth.source + baseline.source_detail = breadth.source + baseline.limit_counts_reliable = breadth.limit_counts_reliable + baseline.data_status = "fresh" if breadth.limit_counts_reliable else "estimated" return baseline, True diff --git a/backend/app/api/debug.py b/backend/app/api/debug.py new file mode 100644 index 00000000..d38e2a53 --- /dev/null +++ b/backend/app/api/debug.py @@ -0,0 +1,289 @@ +"""Debug and operations visibility APIs.""" + +from __future__ import annotations + +import json +import os +from datetime import datetime, timedelta + +from fastapi import APIRouter, Depends +from sqlalchemy import text + +from app.core.deps import get_current_admin +from app.db.database import get_db + +router = APIRouter(prefix="/api/debug", tags=["debug"]) + + +@router.get("/system") +async def get_system_status(_admin: dict = Depends(get_current_admin)): + async with get_db() as db: + errors = await db.execute(text("SELECT COUNT(*) FROM error_logs WHERE created_at >= :start"), {"start": _days_ago(1)}) + scans = await db.execute(text("SELECT COUNT(*) FROM scan_process_logs WHERE created_at >= :start"), {"start": _days_ago(1)}) + latest_scan = await db.execute(text("SELECT * FROM scan_process_logs ORDER BY created_at DESC, id DESC LIMIT 1")) + latest_error = await db.execute(text("SELECT * FROM error_logs ORDER BY created_at DESC, id DESC LIMIT 1")) + error_count = errors.scalar() or 0 + scan_count = scans.scalar() or 0 + return { + "status": "warning" if error_count else "ok", + "generated_at": datetime.now().isoformat(), + "db_size_mb": _db_size_mb(), + "today_error_count": error_count, + "today_scan_log_count": scan_count, + "latest_scan": _row(latest_scan.fetchone()), + "latest_error": _row(latest_error.fetchone()), + } + + +@router.get("/errors") +async def get_error_logs( + limit: int = 50, + source: str | None = None, + level: str | None = None, + days: int = 7, + q: str | None = None, + _admin: dict = Depends(get_current_admin), +): + clauses = ["created_at >= :start"] + params: dict = {"start": _days_ago(days), "limit": min(max(limit, 1), 200)} + if source: + clauses.append("source = :source") + params["source"] = source + if level: + clauses.append("level = :level") + params["level"] = level + if q: + clauses.append("(message LIKE :q OR detail LIKE :q)") + params["q"] = f"%{q}%" + where = " AND ".join(clauses) + async with get_db() as db: + result = await db.execute( + text( + f"SELECT * FROM error_logs WHERE {where} " + "ORDER BY created_at DESC, id DESC LIMIT :limit" + ), + params, + ) + count = await db.execute(text(f"SELECT COUNT(*) FROM error_logs WHERE {where}"), params) + rows = [_row(row) for row in result.fetchall()] + sources = sorted({row.get("source") for row in rows if row.get("source")}) + levels = sorted({row.get("level") for row in rows if row.get("level")}) + return { + "total": count.scalar() or 0, + "errors": rows, + "logs": rows, + "sources": sources, + "levels": levels, + "source_counts": _count_by(rows, "source"), + "level_counts": _count_by(rows, "level"), + } + + +@router.delete("/errors") +async def clear_error_logs(days: int = 30, _admin: dict = Depends(get_current_admin)): + async with get_db() as db: + result = await db.execute( + text("DELETE FROM error_logs WHERE created_at < :start"), + {"start": _days_ago(days)}, + ) + await db.commit() + return {"status": "ok", "deleted": result.rowcount or 0} + + +@router.get("/scan-sessions") +async def get_scan_sessions(days: int = 7, limit: int = 30, _admin: dict = Depends(get_current_admin)): + async with get_db() as db: + result = await db.execute( + text( + "SELECT scan_session, MAX(scan_mode) AS scan_mode, MAX(created_at) AS created_at, " + "COUNT(*) AS stage_count, MAX(status) AS status, MAX(input_count) AS input_count, " + "MAX(output_count) AS final_count, SUM(filtered_count) AS drop_count, MAX(summary) AS last_summary " + "FROM scan_process_logs WHERE created_at >= :start " + "GROUP BY scan_session ORDER BY MAX(created_at) DESC LIMIT :limit" + ), + {"start": _days_ago(days), "limit": min(max(limit, 1), 100)}, + ) + return {"sessions": [_row(row) for row in result.fetchall()]} + + +@router.get("/scan-logs") +async def get_scan_logs(scan_session: str | None = None, days: int = 7, limit: int = 120, _admin: dict = Depends(get_current_admin)): + async with get_db() as db: + if not scan_session: + latest = await db.execute(text("SELECT scan_session FROM scan_process_logs ORDER BY created_at DESC, id DESC LIMIT 1")) + row = latest.fetchone() + scan_session = row._mapping["scan_session"] if row else None + if not scan_session: + return {"scan_session": None, "logs": []} + result = await db.execute( + text( + "SELECT * FROM scan_process_logs WHERE scan_session = :session AND created_at >= :start " + "ORDER BY created_at ASC, id ASC LIMIT :limit" + ), + {"session": scan_session, "start": _days_ago(days), "limit": min(max(limit, 1), 300)}, + ) + return {"scan_session": scan_session, "logs": [_scan_row(row) for row in result.fetchall()]} + + +@router.get("/tasks") +async def get_task_center(_admin: dict = Depends(get_current_admin)): + from app.engine import scheduler as scheduler_module + from app.engine.recommender import _scan_running + + scheduler = getattr(scheduler_module, "scheduler", None) + jobs = [] + if scheduler: + for job in scheduler.get_jobs(): + jobs.append({ + "id": job.id, + "name": job.name, + "next_run_time": str(job.next_run_time or ""), + "trigger": str(job.trigger), + }) + async with get_db() as db: + scan_logs = await db.execute( + text("SELECT * FROM scan_process_logs ORDER BY created_at DESC, id DESC LIMIT 12") + ) + errors = await db.execute( + text("SELECT source, level, message, created_at FROM error_logs ORDER BY created_at DESC, id DESC LIMIT 8") + ) + return { + "scheduler_running": bool(scheduler and scheduler.running), + "scan_running": bool(_scan_running), + "scan_locked": bool(_scan_running), + "job_count": len(jobs), + "jobs": jobs, + "recent_scan_logs": [_row(row) for row in scan_logs.fetchall()], + "recent_errors": [_row(row) for row in errors.fetchall()], + "generated_at": datetime.now().isoformat(), + } + + +@router.get("/data-source-health") +async def get_data_source_health(days: int = 7, _admin: dict = Depends(get_current_admin)): + async with get_db() as db: + result = await db.execute( + text( + "SELECT source, " + "SUM(CASE WHEN level = 'error' THEN 1 ELSE 0 END) AS error_count, " + "SUM(CASE WHEN level = 'warning' THEN 1 ELSE 0 END) AS warning_count, " + "MAX(message) AS last_error, MAX(created_at) AS last_seen_at " + "FROM error_logs WHERE created_at >= :start GROUP BY source ORDER BY error_count DESC, warning_count DESC" + ), + {"start": _days_ago(days)}, + ) + freshness_rows = { + "market_temperature": await _latest_row(db, "market_temperature", "市场温度", "created_at", "trade_date"), + "sector_heat": await _latest_row(db, "sector_heat", "板块主线", "created_at", "trade_date"), + "recommendations": await _latest_row(db, "recommendations", "机会清单", "created_at", "scan_session"), + "research_reports": await _latest_row(db, "research_reports", "研究报告", "created_at", "trade_date"), + "news_items": await _latest_row(db, "news_items", "消息催化", "created_at", "published_at"), + "scan_process_logs": await _latest_row(db, "scan_process_logs", "扫描记录", "created_at", "scan_session"), + } + sources = [] + for row in result.fetchall(): + item = _row(row) + sources.append({ + **item, + "status": "error" if item.get("error_count", 0) else "warning" if item.get("warning_count", 0) else "ok", + }) + freshness = {key: _freshness_status(value) for key, value in freshness_rows.items()} + return {"days": days, "sources": sources, "freshness": freshness, "generated_at": datetime.now().isoformat()} + + +def _days_ago(days: int) -> str: + return (datetime.now() - timedelta(days=max(1, days))).strftime("%Y-%m-%d %H:%M:%S") + + +def _db_size_mb() -> float: + path = os.environ.get("ASTOCK_DATABASE_URL", "").replace("sqlite:///", "") or "./astock.db" + if not os.path.exists(path): + path = "./data/astock.db" + try: + return round(os.path.getsize(path) / 1024 / 1024, 2) + except Exception: + return 0 + + +def _row(row) -> dict: + if not row: + return {} + return {key: _json_safe(value) for key, value in dict(row._mapping).items()} + + +def _scan_row(row) -> dict: + data = _row(row) + raw = data.pop("detail_json", "{}") + try: + data["detail"] = json.loads(raw or "{}") + except Exception: + data["detail"] = {} + return data + + +async def _latest_row(db, table: str, label: str, time_column: str, ref_column: str) -> dict: + result = await db.execute( + text( + f"SELECT MAX({time_column}) AS last_success_at, MAX({ref_column}) AS last_reference, COUNT(*) AS total " + f"FROM {table}" + ) + ) + row = _row(result.fetchone()) + row["label"] = label + row["table"] = table + return row + + +def _freshness_status(item: dict) -> dict: + last_success = item.get("last_success_at") or "" + status = "missing" + age_minutes = None + if last_success: + parsed = _parse_datetime(last_success) + if parsed: + age_minutes = round((datetime.now() - parsed).total_seconds() / 60, 1) + status = "ok" if age_minutes <= 360 else "stale" + return { + **item, + "status": status, + "age_minutes": age_minutes, + "message": _freshness_message(item.get("label", ""), status, age_minutes), + } + + +def _freshness_message(label: str, status: str, age_minutes: float | None) -> str: + if status == "missing": + return f"{label}暂无成功记录" + if age_minutes is None: + return f"{label}时间无法识别" + if age_minutes <= 60: + return f"{label}刚更新" + if age_minutes <= 360: + return f"{label}{round(age_minutes / 60, 1)}小时前更新" + return f"{label}超过6小时未更新" + + +def _parse_datetime(value: str): + if isinstance(value, datetime): + return value + for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d"): + try: + return datetime.strptime(str(value).split(".")[0], fmt) + except Exception: + continue + return None + + +def _json_safe(value): + if isinstance(value, datetime): + return value.isoformat() + return value + + +def _count_by(rows: list[dict], key: str) -> dict: + counts: dict[str, int] = {} + for row in rows: + value = row.get(key) + if value: + counts[str(value)] = counts.get(str(value), 0) + 1 + return counts diff --git a/backend/app/api/research.py b/backend/app/api/research.py new file mode 100644 index 00000000..9e3a06e2 --- /dev/null +++ b/backend/app/api/research.py @@ -0,0 +1,264 @@ +"""Research report APIs.""" + +from __future__ import annotations + +import json +from datetime import datetime, timedelta + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel, Field +from sqlalchemy import text + +from app.core.deps import get_current_admin +from app.db import tables +from app.db.database import get_db +from app.research.industry_chain_agent import ensure_theme_knowledge_seeded, load_theme_chain_library +from app.research.report_agent import load_latest_research_report + +router = APIRouter(prefix="/api/research", tags=["research"]) + + +class ThemeKnowledgeUpdate(BaseModel): + theme_name: str = Field(min_length=1) + aliases: list[str] = Field(default_factory=list) + logic: str = "" + lifecycle_status: str = "观察期" + stage: str = "mid" + chain_nodes: list[str] = Field(default_factory=list) + chain_items: list[dict] = Field(default_factory=list) + is_active: bool = True + sort_order: int = 0 + + +@router.get("/today") +async def get_today_research(): + report = await load_latest_research_report() + if report: + return report + from app.engine.recommender import get_latest_recommendations + from app.research.report_agent import build_research_report_async + + latest = await get_latest_recommendations() + latest_scan = latest.get("latest_scan") or {} + if latest_scan: + return await build_research_report_async(latest, latest_scan.get("scan_session") or "latest") + return { + "trade_date": datetime.now().strftime("%Y%m%d"), + "scan_session": "", + "scanned_at": "", + "market_view": {"regime": "unknown", "confidence": 0, "summary": "暂无研究报告。"}, + "theme_views": [], + "industry_chain_map": [], + "opportunity_cards": [], + "risk_alerts": [], + "no_trade_reason": {"has_scan": False, "reason": "暂无完成扫描。"}, + } + + +@router.get("/history") +async def get_research_history(days: int = 14): + start = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d") + async with get_db() as db: + result = await db.execute( + text( + "SELECT scan_session, trade_date, market_summary, theme_summary, no_trade_reason, report_json, created_at " + "FROM research_reports WHERE created_at >= :start " + "ORDER BY created_at DESC, id DESC LIMIT 60" + ), + {"start": start}, + ) + rows = [] + for row in result.fetchall(): + r = row._mapping + rows.append({ + "scan_session": r["scan_session"], + "trade_date": r["trade_date"], + "market_summary": r["market_summary"] or "", + "theme_summary": r["theme_summary"] or "", + "no_trade_reason": _safe_json(r["no_trade_reason"]), + "created_at": str(r["created_at"] or ""), + }) + return rows + + +@router.get("/themes") +async def get_research_themes(): + async with get_db() as db: + result = await db.execute( + text( + "SELECT * FROM theme_maps " + "WHERE scan_session = (SELECT scan_session FROM research_reports ORDER BY created_at DESC, id DESC LIMIT 1) " + "ORDER BY heat_score DESC" + ) + ) + return [dict(row._mapping) for row in result.fetchall()] + + +@router.get("/opportunities") +async def get_research_opportunities(): + async with get_db() as db: + result = await db.execute( + text( + "SELECT * FROM opportunity_cards " + "WHERE scan_session = (SELECT scan_session FROM research_reports ORDER BY created_at DESC, id DESC LIMIT 1) " + "ORDER BY score DESC" + ) + ) + return [dict(row._mapping) for row in result.fetchall()] + + +@router.get("/risks") +async def get_research_risks(): + async with get_db() as db: + result = await db.execute( + text( + "SELECT * FROM risk_events " + "WHERE scan_session = (SELECT scan_session FROM research_reports ORDER BY created_at DESC, id DESC LIMIT 1) " + "ORDER BY reject DESC, severity DESC, id DESC" + ) + ) + return [dict(row._mapping) for row in result.fetchall()] + + +@router.get("/review") +async def get_research_review(days: int = 60): + from app.research.review_agent import build_research_review + + return await build_research_review(days=days) + + +@router.get("/theme-knowledge") +async def get_theme_knowledge(): + return await load_theme_chain_library() + + +@router.put("/theme-knowledge/{theme_name}") +async def update_theme_knowledge( + theme_name: str, + payload: ThemeKnowledgeUpdate, + _admin: dict = Depends(get_current_admin), +): + await ensure_theme_knowledge_seeded() + normalized_name = payload.theme_name.strip() or theme_name.strip() + if not normalized_name: + raise HTTPException(status_code=400, detail="主题名称不能为空") + + chain_nodes = [node.strip() for node in payload.chain_nodes if node.strip()] + if payload.chain_items: + chain_nodes = [str(item.get("chain_node") or "").strip() for item in payload.chain_items if str(item.get("chain_node") or "").strip()] + if not chain_nodes: + raise HTTPException(status_code=400, detail="至少需要一个产业链环节") + + async with get_db() as db: + existing = await db.execute( + text("SELECT id FROM theme_knowledge WHERE theme_name = :theme_name LIMIT 1"), + {"theme_name": theme_name}, + ) + row = existing.fetchone() + values = { + "theme_name": normalized_name, + "aliases_json": json.dumps([item.strip() for item in payload.aliases if item.strip()], ensure_ascii=False), + "logic_summary": payload.logic.strip(), + "lifecycle_status": payload.lifecycle_status.strip() or "观察期", + "stage": payload.stage.strip() or "mid", + "is_active": bool(payload.is_active), + "sort_order": int(payload.sort_order or 0), + } + if row: + await db.execute( + text( + "UPDATE theme_knowledge SET theme_name = :theme_name, aliases_json = :aliases_json, " + "logic_summary = :logic_summary, lifecycle_status = :lifecycle_status, stage = :stage, " + "is_active = :is_active, sort_order = :sort_order, updated_at = CURRENT_TIMESTAMP " + "WHERE id = :id" + ), + {**values, "id": row._mapping["id"]}, + ) + else: + await db.execute(tables.theme_knowledge_table.insert().values(**values)) + + await db.execute(text("DELETE FROM theme_chain_knowledge WHERE theme_name = :theme_name"), {"theme_name": theme_name}) + if normalized_name != theme_name: + await db.execute(text("DELETE FROM theme_chain_knowledge WHERE theme_name = :theme_name"), {"theme_name": normalized_name}) + await db.execute( + tables.theme_chain_knowledge_table.insert(), + [ + { + "theme_name": normalized_name, + "chain_node": item["chain_node"], + "related_stocks": json.dumps(item.get("related_stocks", []), ensure_ascii=False, default=str), + "leader_stocks": json.dumps(item.get("leader_stocks", []), ensure_ascii=False, default=str), + "node_role": item.get("node_role", ""), + "is_active": True, + "sort_order": index, + } + for index, item in enumerate(_normalize_chain_items(payload.chain_items, chain_nodes)) + ], + ) + await db.commit() + + library = await load_theme_chain_library() + for item in library: + if item.get("theme") == normalized_name: + return item + return {"status": "ok", "theme": normalized_name} + + +@router.post("/refresh") +async def refresh_research(_admin: dict = Depends(get_current_admin)): + from app.engine.recommender import get_latest_recommendations + from app.research.report_agent import build_research_report_async, save_research_report + + result = await get_latest_recommendations() + latest_scan = result.get("latest_scan") or {} + scan_session = latest_scan.get("scan_session") or "manual_research" + report = await build_research_report_async(result, scan_session) + await save_research_report(report) + return {"status": "ok", "scan_session": scan_session, "opportunity_count": len(report.get("opportunity_cards", []))} + + +def _safe_json(value: str | None) -> dict: + if not value: + return {} + try: + parsed = json.loads(value) + return parsed if isinstance(parsed, dict) else {} + except Exception: + return {} + + +def _normalize_chain_items(chain_items: list[dict], chain_nodes: list[str]) -> list[dict]: + if not chain_items: + return [ + {"chain_node": node, "related_stocks": [], "leader_stocks": [], "node_role": ""} + for node in chain_nodes + ] + normalized = [] + for item in chain_items: + node = str(item.get("chain_node") or "").strip() + if not node: + continue + normalized.append({ + "chain_node": node, + "related_stocks": _stock_list(item.get("related_stocks")), + "leader_stocks": _stock_list(item.get("leader_stocks")), + "node_role": str(item.get("node_role") or "").strip(), + }) + return normalized + + +def _stock_list(value) -> list: + if not isinstance(value, list): + return [] + cleaned = [] + for item in value: + if isinstance(item, dict): + name = str(item.get("name") or "").strip() + ts_code = str(item.get("ts_code") or item.get("code") or "").strip() + if name or ts_code: + cleaned.append({"name": name, "ts_code": ts_code}) + else: + text_value = str(item).strip() + if text_value: + cleaned.append(text_value) + return cleaned diff --git a/backend/app/config.py b/backend/app/config.py index fd8ce8a7..9c7e9e08 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -89,6 +89,24 @@ class Settings(BaseSettings): recommendation_push_max_items: int = 8 recommendation_push_dedup_ttl_seconds: int = 600 + # 研究层风险 Agent + research_risk_enabled: bool = True + research_risk_stock_limit: int = 12 + risk_unlock_lookahead_days: int = 45 + risk_holder_trade_lookback_days: int = 120 + risk_forecast_lookback_days: int = 210 + risk_announcement_lookback_days: int = 180 + risk_financial_lookback_days: int = 540 + risk_unlock_reject_ratio: float = 15.0 + risk_pledge_reject_ratio: float = 55.0 + risk_goodwill_assets_warning_ratio: float = 20.0 + risk_goodwill_assets_reject_ratio: float = 35.0 + risk_debt_assets_warning_ratio: float = 70.0 + risk_debt_assets_reject_ratio: float = 85.0 + research_stock_llm_enabled: bool = True + research_stock_llm_limit: int = 8 + research_stock_news_limit: int = 6 + # 前端 frontend_url: str = "http://localhost:3002" diff --git a/backend/app/data/models.py b/backend/app/data/models.py index b0ecf18d..586d85d0 100644 --- a/backend/app/data/models.py +++ b/backend/app/data/models.py @@ -102,6 +102,10 @@ class MarketTemperature(BaseModel): broken_rate: float = 0 # 炸板率 % index_above_ma20: bool = False # 上证在 MA20 上方 temperature: float = 0 # 综合温度 0-100 + source: str = "snapshot" + data_status: str = "fresh" # fresh / estimated / degraded / snapshot + source_detail: str = "" + limit_counts_reliable: bool = False class MarketBreadth(BaseModel): diff --git a/backend/app/data/tushare_client.py b/backend/app/data/tushare_client.py index 2f16a5c3..aa5efb30 100644 --- a/backend/app/data/tushare_client.py +++ b/backend/app/data/tushare_client.py @@ -266,6 +266,127 @@ class TushareClient: ) ) + # ── 风险事件：解禁 / 减持 / 业绩预告 / 质押 ── + + def get_share_float( + self, + ts_code: str, + start_date: str, + end_date: str, + ) -> pd.DataFrame: + """获取限售股解禁/流通股本变动信息。""" + key = f"share_float:{ts_code}:{start_date}:{end_date}" + return self._cached_fetch( + key, settings.cache_ttl_daily, + lambda: self.pro.share_float( + ts_code=ts_code, + start_date=start_date, + end_date=end_date, + fields="ts_code,ann_date,float_date,float_share,float_ratio,holder_name,share_type" + ) + ) + + def get_holder_trade( + self, + ts_code: str, + start_date: str, + end_date: str, + ) -> pd.DataFrame: + """获取股东增减持信息。""" + key = f"holder_trade:{ts_code}:{start_date}:{end_date}" + return self._cached_fetch( + key, settings.cache_ttl_daily, + lambda: self.pro.stk_holdertrade( + ts_code=ts_code, + start_date=start_date, + end_date=end_date, + fields="ts_code,ann_date,holder_name,holder_type,in_de,change_vol,change_ratio,after_share,after_ratio,avg_price,total_share,begin_date,close_date" + ) + ) + + def get_forecast( + self, + ts_code: str, + start_date: str, + end_date: str, + ) -> pd.DataFrame: + """获取业绩预告。""" + key = f"forecast:{ts_code}:{start_date}:{end_date}" + return self._cached_fetch( + key, settings.cache_ttl_daily, + lambda: self.pro.forecast( + ts_code=ts_code, + start_date=start_date, + end_date=end_date, + fields="ts_code,ann_date,type,p_change_min,p_change_max,net_profit_min,net_profit_max,summary,change_reason" + ) + ) + + def get_pledge_stat(self, ts_code: str) -> pd.DataFrame: + """获取股权质押统计。""" + key = f"pledge_stat:{ts_code}" + return self._cached_fetch( + key, settings.cache_ttl_daily, + lambda: self.pro.pledge_stat( + ts_code=ts_code, + fields="ts_code,end_date,pledge_count,unrest_pledge,rest_pledge,total_share,pledge_ratio" + ) + ) + + def get_announcements( + self, + ts_code: str, + start_date: str, + end_date: str, + ) -> pd.DataFrame: + """获取上市公司公告，用于扫描监管、诉讼、处罚等重大风险标题。""" + key = f"anns_d:{ts_code}:{start_date}:{end_date}" + return self._cached_fetch( + key, settings.cache_ttl_daily, + lambda: self.pro.anns_d( + ts_code=ts_code, + start_date=start_date, + end_date=end_date, + fields="ts_code,ann_date,ann_time,title,url" + ) + ) + + def get_fina_audit( + self, + ts_code: str, + start_date: str, + end_date: str, + ) -> pd.DataFrame: + """获取财务审计意见。""" + key = f"fina_audit:{ts_code}:{start_date}:{end_date}" + return self._cached_fetch( + key, settings.cache_ttl_daily, + lambda: self.pro.fina_audit( + ts_code=ts_code, + start_date=start_date, + end_date=end_date, + fields="ts_code,ann_date,end_date,audit_result,audit_fees,audit_agency" + ) + ) + + def get_balance_sheet( + self, + ts_code: str, + start_date: str, + end_date: str, + ) -> pd.DataFrame: + """获取资产负债表关键风险字段。""" + key = f"balancesheet:{ts_code}:{start_date}:{end_date}" + return self._cached_fetch( + key, settings.cache_ttl_daily, + lambda: self.pro.balancesheet( + ts_code=ts_code, + start_date=start_date, + end_date=end_date, + fields="ts_code,ann_date,end_date,total_assets,total_liab,goodwill" + ) + ) + # ── 新闻快讯 ── def get_news( diff --git a/backend/app/db/database.py b/backend/app/db/database.py index f556f3ff..d637fe25 100644 --- a/backend/app/db/database.py +++ b/backend/app/db/database.py @@ -101,6 +101,33 @@ async def init_db(): "ALTER TABLE news_items ADD COLUMN summary TEXT DEFAULT ''", "ALTER TABLE news_items ADD COLUMN error TEXT DEFAULT ''", "ALTER TABLE catalysts ADD COLUMN llm_reason TEXT DEFAULT ''", + "ALTER TABLE stock_research_notes ADD COLUMN disagreement TEXT DEFAULT ''", + "ALTER TABLE stock_research_notes ADD COLUMN invalid_condition TEXT DEFAULT ''", + "ALTER TABLE stock_research_notes ADD COLUMN generated_by TEXT DEFAULT 'rules'", + "ALTER TABLE stock_research_notes ADD COLUMN stock_role TEXT DEFAULT '待归类'", + "ALTER TABLE stock_research_notes ADD COLUMN theme TEXT DEFAULT '未归类'", + "ALTER TABLE stock_research_notes ADD COLUMN chain_node TEXT DEFAULT '未归类'", + "ALTER TABLE opportunity_cards ADD COLUMN stock_role TEXT DEFAULT '待归类'", + "ALTER TABLE opportunity_cards ADD COLUMN alpha_type TEXT DEFAULT '观察线索'", + "ALTER TABLE opportunity_cards ADD COLUMN alpha_score REAL DEFAULT 0", + "ALTER TABLE opportunity_cards ADD COLUMN beta_dependency TEXT DEFAULT '中'", + "ALTER TABLE opportunity_cards ADD COLUMN beta_dependency_score REAL DEFAULT 0", + "ALTER TABLE opportunity_cards ADD COLUMN ambush_score REAL DEFAULT 0", + "ALTER TABLE opportunity_cards ADD COLUMN expectation_gap_score REAL DEFAULT 0", + "ALTER TABLE opportunity_cards ADD COLUMN risk_gate TEXT DEFAULT '通过'", + "ALTER TABLE opportunity_cards ADD COLUMN setup_quality TEXT DEFAULT '仅观察'", + "ALTER TABLE opportunity_cards ADD COLUMN alpha_reason TEXT DEFAULT ''", + "ALTER TABLE theme_knowledge ADD COLUMN lifecycle_status TEXT DEFAULT '观察期'", + "ALTER TABLE theme_knowledge ADD COLUMN stage TEXT DEFAULT 'mid'", + "ALTER TABLE theme_knowledge ADD COLUMN is_active BOOLEAN DEFAULT 1", + "ALTER TABLE theme_knowledge ADD COLUMN sort_order INTEGER DEFAULT 0", + "ALTER TABLE theme_knowledge ADD COLUMN updated_at DATETIME DEFAULT CURRENT_TIMESTAMP", + "ALTER TABLE theme_chain_knowledge ADD COLUMN related_stocks TEXT DEFAULT '[]'", + "ALTER TABLE theme_chain_knowledge ADD COLUMN leader_stocks TEXT DEFAULT '[]'", + "ALTER TABLE theme_chain_knowledge ADD COLUMN node_role TEXT DEFAULT ''", + "ALTER TABLE theme_chain_knowledge ADD COLUMN is_active BOOLEAN DEFAULT 1", + "ALTER TABLE theme_chain_knowledge ADD COLUMN sort_order INTEGER DEFAULT 0", + "ALTER TABLE theme_chain_knowledge ADD COLUMN updated_at DATETIME DEFAULT CURRENT_TIMESTAMP", ]: try: await conn.execute( @@ -113,11 +140,21 @@ async def init_db(): "CREATE UNIQUE INDEX IF NOT EXISTS idx_news_items_dedup_key ON news_items(dedup_key)", "CREATE INDEX IF NOT EXISTS idx_news_items_status_time ON news_items(status, published_at)", "CREATE INDEX IF NOT EXISTS idx_catalysts_source_url ON catalysts(source, url)", + "CREATE UNIQUE INDEX IF NOT EXISTS idx_theme_knowledge_name ON theme_knowledge(theme_name)", + "CREATE INDEX IF NOT EXISTS idx_theme_knowledge_active_order ON theme_knowledge(is_active, sort_order)", + "CREATE INDEX IF NOT EXISTS idx_theme_chain_knowledge_theme ON theme_chain_knowledge(theme_name, is_active, sort_order)", "CREATE INDEX IF NOT EXISTS idx_scan_process_session_time ON scan_process_logs(scan_session, created_at)", "CREATE INDEX IF NOT EXISTS idx_scan_process_stage_time ON scan_process_logs(stage, created_at)", "CREATE INDEX IF NOT EXISTS idx_research_observations_session_score ON research_observations(scan_session, final_score)", "CREATE INDEX IF NOT EXISTS idx_research_observations_code_time ON research_observations(ts_code, created_at)", "CREATE INDEX IF NOT EXISTS idx_research_observations_theme_time ON research_observations(theme_name, created_at)", + "CREATE INDEX IF NOT EXISTS idx_research_reports_session_time ON research_reports(scan_session, created_at)", + "CREATE INDEX IF NOT EXISTS idx_research_reports_trade_date ON research_reports(trade_date, created_at)", + "CREATE INDEX IF NOT EXISTS idx_theme_maps_trade_date_score ON theme_maps(trade_date, heat_score)", + "CREATE INDEX IF NOT EXISTS idx_theme_chain_theme ON theme_chain_nodes(theme_name, trade_date)", + "CREATE INDEX IF NOT EXISTS idx_stock_research_code_time ON stock_research_notes(ts_code, created_at)", + "CREATE INDEX IF NOT EXISTS idx_risk_events_session_reject ON risk_events(scan_session, reject)", + "CREATE INDEX IF NOT EXISTS idx_opportunity_cards_session_score ON opportunity_cards(scan_session, score)", ]: try: await conn.execute(__import__("sqlalchemy").text(index_sql)) diff --git a/backend/app/db/tables.py b/backend/app/db/tables.py index 8bce7662..7ab38fff 100644 --- a/backend/app/db/tables.py +++ b/backend/app/db/tables.py @@ -323,3 +323,129 @@ theme_catalysts_table = Table( Column("reason", Text, default=""), Column("created_at", DateTime, server_default=func.now()), ) + +theme_knowledge_table = Table( + "theme_knowledge", metadata, + Column("id", Integer, primary_key=True, autoincrement=True), + Column("theme_name", Text, nullable=False), + Column("aliases_json", Text, default="[]"), + Column("logic_summary", Text, default=""), + Column("lifecycle_status", Text, default="观察期"), + Column("stage", Text, default="mid"), + Column("is_active", Boolean, default=True), + Column("sort_order", Integer, default=0), + Column("created_at", DateTime, server_default=func.now()), + Column("updated_at", DateTime, server_default=func.now()), +) + +theme_chain_knowledge_table = Table( + "theme_chain_knowledge", metadata, + Column("id", Integer, primary_key=True, autoincrement=True), + Column("theme_name", Text, nullable=False), + Column("chain_node", Text, nullable=False), + Column("related_stocks", Text, default="[]"), + Column("leader_stocks", Text, default="[]"), + Column("node_role", Text, default=""), + Column("is_active", Boolean, default=True), + Column("sort_order", Integer, default=0), + Column("created_at", DateTime, server_default=func.now()), + Column("updated_at", DateTime, server_default=func.now()), +) + +research_reports_table = Table( + "research_reports", metadata, + Column("id", Integer, primary_key=True, autoincrement=True), + Column("scan_session", Text, nullable=False), + Column("trade_date", Text, nullable=False), + Column("market_summary", Text, default=""), + Column("theme_summary", Text, default=""), + Column("no_trade_reason", Text, default=""), + Column("report_json", Text, default="{}"), + Column("created_at", DateTime, server_default=func.now()), +) + +theme_maps_table = Table( + "theme_maps", metadata, + Column("id", Integer, primary_key=True, autoincrement=True), + Column("scan_session", Text, nullable=False), + Column("trade_date", Text, nullable=False), + Column("theme_name", Text, nullable=False), + Column("stage", Text, default="mid"), + Column("heat_score", Float, default=0), + Column("logic_summary", Text, default=""), + Column("lifecycle_status", Text, default=""), + Column("created_at", DateTime, server_default=func.now()), +) + +theme_chain_nodes_table = Table( + "theme_chain_nodes", metadata, + Column("id", Integer, primary_key=True, autoincrement=True), + Column("scan_session", Text, nullable=False), + Column("trade_date", Text, nullable=False), + Column("theme_name", Text, nullable=False), + Column("chain_node", Text, nullable=False), + Column("related_stocks", Text, default="[]"), + Column("leader_stocks", Text, default="[]"), + Column("created_at", DateTime, server_default=func.now()), +) + +stock_research_notes_table = Table( + "stock_research_notes", metadata, + Column("id", Integer, primary_key=True, autoincrement=True), + Column("scan_session", Text, nullable=False), + Column("trade_date", Text, nullable=False), + Column("ts_code", Text, nullable=False), + Column("name", Text, nullable=False), + Column("theme", Text, default="未归类"), + Column("chain_node", Text, default="未归类"), + Column("logic_score", Float, default=0), + Column("logic_summary", Text, default=""), + Column("evidence_json", Text, default="[]"), + Column("uncertainty", Text, default=""), + Column("stock_role", Text, default="待归类"), + Column("disagreement", Text, default=""), + Column("invalid_condition", Text, default=""), + Column("generated_by", Text, default="rules"), + Column("created_at", DateTime, server_default=func.now()), +) + +risk_events_table = Table( + "risk_events", metadata, + Column("id", Integer, primary_key=True, autoincrement=True), + Column("scan_session", Text, nullable=False), + Column("trade_date", Text, nullable=False), + Column("ts_code", Text, default=""), + Column("risk_type", Text, nullable=False), + Column("severity", Text, default="warning"), + Column("reject", Boolean, default=False), + Column("reason", Text, default=""), + Column("source", Text, default="research_agent"), + Column("created_at", DateTime, server_default=func.now()), +) + +opportunity_cards_table = Table( + "opportunity_cards", metadata, + Column("id", Integer, primary_key=True, autoincrement=True), + Column("scan_session", Text, nullable=False), + Column("trade_date", Text, nullable=False), + Column("ts_code", Text, nullable=False), + Column("name", Text, nullable=False), + Column("theme", Text, default=""), + Column("chain_node", Text, default="未归类"), + Column("stock_role", Text, default="待归类"), + Column("opportunity_type", Text, default="观察"), + Column("score", Float, default=0), + Column("alpha_type", Text, default="观察线索"), + Column("alpha_score", Float, default=0), + Column("beta_dependency", Text, default="中"), + Column("beta_dependency_score", Float, default=0), + Column("ambush_score", Float, default=0), + Column("expectation_gap_score", Float, default=0), + Column("risk_gate", Text, default="通过"), + Column("setup_quality", Text, default="仅观察"), + Column("alpha_reason", Text, default=""), + Column("action_plan", Text, default="观察"), + Column("trigger", Text, default=""), + Column("invalid_condition", Text, default=""), + Column("created_at", DateTime, server_default=func.now()), +) diff --git a/backend/app/engine/recommender.py b/backend/app/engine/recommender.py index 6bb81420..547525c9 100644 --- a/backend/app/engine/recommender.py +++ b/backend/app/engine/recommender.py @@ -163,6 +163,9 @@ async def refresh_recommendations(trade_date: str = None, scan_session: str = "m # 持久化到数据库（这是 async 操作，需要在主线程中执行） await _save_to_db(result) + # 生成 AI 研究报告骨架：不改变推荐算法，只把扫描结果升级为研究产物。 + await _save_research_layer(result, scan_session) + # 推送本轮可操作/重点关注推荐，失败不影响扫描结果。 await _push_recommendation_notifications(result, scan_session) @@ -190,6 +193,42 @@ async def _push_recommendation_notifications(result: dict, scan_session: str) -> logger.warning("飞书推荐推送失败: %s", e) +async def _save_research_layer(result: dict, scan_session: str) -> None: + try: + from app.research.report_agent import build_research_report_async, save_research_report + from sqlalchemy import text + + if not result.get("latest_scan"): + async with get_db() as db: + scan_row = await db.execute( + text( + "SELECT * FROM scan_process_logs " + "WHERE scan_session = :session AND stage = 'final_filter' " + "ORDER BY created_at DESC, id DESC LIMIT 1" + ), + {"session": scan_session}, + ) + result["latest_scan"] = _scan_meta_from_row(scan_row.fetchone()) + report = await build_research_report_async(result, scan_session) + await save_research_report(report) + try: + from app.notifications.feishu import send_research_report_push + + await send_research_report_push(report) + except Exception as push_error: + logger.warning("研究日报推送失败: %s", push_error) + logger.info( + "已生成研究报告: scan_session=%s opportunities=%s risks=%s", + scan_session, + len(report.get("opportunity_cards", [])), + len(report.get("risk_alerts", [])), + ) + except Exception as e: + logger.warning("研究报告生成失败: %s", e) + from app.db.error_logger import log_error + await log_error("research", f"研究报告生成失败: {e}", detail=traceback.format_exc()) + + async def _update_tracking(): """更新历史推荐的跟踪数据""" try: @@ -851,6 +890,7 @@ async def _save_to_db(result: dict): # 保存市场温度 mt = result.get("market_temp") if mt: + _calibrate_market_temp_for_persistence(mt, result.get("hot_sectors", []) or []) if _has_valid_market_breadth(mt): # 使用 INSERT OR REPLACE 确保重复扫描能更新数据 stmt = text( @@ -988,6 +1028,22 @@ async def _save_to_db(result: dict): await log_error("recommender", f"保存推荐到数据库失败: {e}", detail=traceback.format_exc()) +def _calibrate_market_temp_for_persistence(market_temp, hot_sectors: list) -> None: + if not market_temp or not hot_sectors: + return + sector_limit_up = sum(max(int(getattr(sector, "limit_up_count", 0) or 0), 0) for sector in hot_sectors) + if sector_limit_up <= int(getattr(market_temp, "limit_up_count", 0) or 0): + return + original_limit = int(getattr(market_temp, "limit_up_count", 0) or 0) + original_temp = float(getattr(market_temp, "temperature", 0) or 0) + market_temp.limit_up_count = sector_limit_up + if original_limit == 0: + market_temp.temperature = round(min(original_temp + min(sector_limit_up / 5, 8), 100), 1) + if hasattr(market_temp, "data_status") and not getattr(market_temp, "limit_counts_reliable", False): + market_temp.data_status = "estimated" + market_temp.source_detail = f"{getattr(market_temp, 'source_detail', '')};sector_limit_lower_bound" + + async def _load_today_from_db() -> dict: """从数据库加载今日推荐""" today = datetime.now().strftime("%Y-%m-%d") @@ -1113,10 +1169,11 @@ async def _load_today_from_db() -> dict: strategy_profile = await _load_latest_strategy_profile_for_session(db, latest_scan_session) if not strategy_profile and recommendations: strategy_profile = get_strategy_profile_by_id(recommendations[0].strategy).model_dump() + latest_sectors = await _load_sectors_from_db() return { "market_temp": market_temp, - "hot_sectors": [], + "hot_sectors": latest_sectors, "capital_filtered": [], "recommendations": recommendations, "strategy_profile": strategy_profile, diff --git a/backend/app/engine/screener.py b/backend/app/engine/screener.py index d7a12190..59b787c9 100644 --- a/backend/app/engine/screener.py +++ b/backend/app/engine/screener.py @@ -155,6 +155,20 @@ async def run_screening(trade_date: str = None, scan_session: str = "manual") -> merge_sectors_to_themes(await intraday_sector_scan(hot_sectors), limit=settings.top_sector_count) ) + calibration = _calibrate_market_temperature_from_sectors(market_temp, hot_sectors) + if calibration: + await log_scan_stage( + scan_session=scan_session, + scan_mode=scan_mode, + stage="market_breadth_calibration", + stage_label="市场情绪校准", + input_count=len(hot_sectors), + output_count=1, + filtered_count=0, + summary=calibration["summary"], + detail=calibration, + ) + strategy_profile = await select_strategy_profile(market_temp, hot_sectors, intraday, scan_session=scan_session) logger.info( f"=== 今日策略: {strategy_profile.name} ({strategy_profile.strategy_id}) " @@ -792,6 +806,41 @@ def _route_recall_weight(route: str, item: dict) -> float: return 0 +def _calibrate_market_temperature_from_sectors( + market_temp: MarketTemperature, + hot_sectors: list[SectorInfo], +) -> dict | None: + """Use theme limit-up counts as a lower bound when market breadth limits are missing.""" + if not market_temp or not hot_sectors: + return None + sector_limit_up = sum(max(int(getattr(sector, "limit_up_count", 0) or 0), 0) for sector in hot_sectors) + sector_limit_down = sum(max(int(getattr(sector, "realtime_down_count", 0) or 0), 0) for sector in hot_sectors) + if sector_limit_up <= 0: + return None + original_limit_up = int(market_temp.limit_up_count or 0) + original_temp = float(market_temp.temperature or 0) + if original_limit_up >= sector_limit_up: + return None + + market_temp.limit_up_count = sector_limit_up + if sector_limit_down and not market_temp.limit_down_count: + market_temp.limit_down_count = sector_limit_down + if not getattr(market_temp, "limit_counts_reliable", False): + market_temp.data_status = "estimated" + market_temp.source_detail = f"{getattr(market_temp, 'source_detail', '')};sector_limit_lower_bound" + if original_limit_up == 0: + market_temp.temperature = round(min(original_temp + min(sector_limit_up / 5, 8), 100), 1) + + return { + "summary": f"全市场涨停数由 {original_limit_up} 校准为至少 {sector_limit_up}，市场温度 {original_temp:.1f} -> {market_temp.temperature:.1f}", + "original_limit_up_count": original_limit_up, + "calibrated_limit_up_count": sector_limit_up, + "original_temperature": original_temp, + "calibrated_temperature": market_temp.temperature, + "source": "sector_limit_lower_bound", + } + + def _finalize_battle_plan( recommendations: list[Recommendation], hot_sectors: list[SectorInfo], diff --git a/backend/app/main.py b/backend/app/main.py index f4b8500f..319716df 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -11,7 +11,7 @@ from app.config import settings from app.db.error_logger import PersistentErrorLogHandler, log_error from app.db.database import init_db from app.engine.scheduler import start_scheduler, stop_scheduler -from app.api import market, sectors, recommendations, stocks, watchlists, websocket, chat, auth, catalysts +from app.api import market, sectors, recommendations, stocks, watchlists, websocket, chat, auth, catalysts, research, debug def configure_logging() -> None: logging.basicConfig( @@ -145,6 +145,8 @@ app.include_router(watchlists.router) app.include_router(chat.router) app.include_router(auth.router) app.include_router(catalysts.router) +app.include_router(research.router) +app.include_router(debug.router) # WebSocket app.websocket("/ws")(websocket.ws_endpoint) diff --git a/backend/app/notifications/feishu.py b/backend/app/notifications/feishu.py index 08d0fe93..aa6ea04b 100644 --- a/backend/app/notifications/feishu.py +++ b/backend/app/notifications/feishu.py @@ -298,3 +298,92 @@ async def send_recommendation_push( except Exception as e: logger.warning("Feishu 推荐推送失败: %s", e) return False + + +def _research_signature(report: dict) -> str: + basis = "|".join([ + str(report.get("scan_session", "")), + str(report.get("trade_date", "")), + str(len(report.get("opportunity_cards", []) or [])), + str((report.get("no_trade_reason") or {}).get("reason", "")), + ]) + return hashlib.sha1(basis.encode("utf-8")).hexdigest() + + +def _build_research_card(report: dict, now: str) -> dict: + market = report.get("market_view") or {} + themes = report.get("theme_views") or [] + opportunities = report.get("opportunity_cards") or [] + risks = report.get("risk_alerts") or [] + no_trade = report.get("no_trade_reason") or {} + header_template = "red" if opportunities else "orange" + + summary = "\n".join([ + f"**扫描**: {report.get('scan_session') or '-'}", + f"**时间**: {now}", + f"**市场状态**: {market.get('summary') or '-'}", + f"**机会卡**: {len(opportunities)} 个 **风险**: {len(risks)} 个", + ]) + elements: list[dict] = [{"tag": "div", "text": _card_text(summary)}, {"tag": "hr"}] + + if themes: + theme_lines = [] + for item in themes[:3]: + nodes = " / ".join((item.get("chain_nodes") or [])[:4]) + theme_lines.append(f"**{item.get('theme')}** {item.get('heat_score', 0):.0f}分 · {item.get('lifecycle_status') or item.get('stage')} · {nodes}") + elements.append({"tag": "div", "text": _card_text("**主线/产业链**\n" + "\n".join(theme_lines))}) + + if opportunities: + elements.append({"tag": "hr"}) + lines = [] + for item in opportunities[:5]: + lines.append(f"**{item.get('name')}** `{item.get('ts_code')}` · {item.get('opportunity_type')} · {item.get('theme')}/{item.get('chain_node')} · {item.get('score')}分") + elements.append({"tag": "div", "text": _card_text("**Top 机会**\n" + "\n".join(lines))}) + else: + elements.append({"tag": "hr"}) + elements.append({"tag": "div", "text": _card_text(f"**本轮无交易级机会**\n{no_trade.get('reason') or '没有形成满足条件的机会卡。'}")}) + + if risks: + elements.append({"tag": "hr"}) + risk_lines = [f"{'否决' if item.get('reject') else '预警'} · {item.get('reason')}" for item in risks[:3]] + elements.append({"tag": "div", "text": _card_text("**风险雷达**\n" + "\n".join(risk_lines))}) + + elements.append({ + "tag": "note", + "elements": [_card_text("研究日报用于解释市场、主线和风险；交易仍需等待触发条件确认。", tag="plain_text")], + }) + return { + "config": {"wide_screen_mode": True}, + "header": { + "template": header_template, + "title": _card_text(f"{settings.alert_app_name} 研究日报", tag="plain_text"), + }, + "elements": elements, + } + + +async def send_research_report_push(report: dict) -> bool: + webhook_url = settings.recommendation_push_webhook_url or settings.feishu_webhook_url + if not settings.recommendation_push_enabled or not webhook_url: + return False + + signature = _research_signature(report) + dedup_key = f"feishu_research_report:{signature}" + if cache.get(dedup_key): + return False + cache.set(dedup_key, True, settings.recommendation_push_dedup_ttl_seconds) + + now = datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y-%m-%d %H:%M:%S") + payload = {"msg_type": "interactive", "card": _build_research_card(report, now)} + try: + async with httpx.AsyncClient(timeout=8, follow_redirects=True) as client: + resp = await client.post(webhook_url, json=payload) + resp.raise_for_status() + body = resp.json() + if body.get("code", 0) != 0: + logger.warning("Feishu 研究日报推送失败: %s", body) + return False + return True + except Exception as e: + logger.warning("Feishu 研究日报推送失败: %s", e) + return False diff --git a/backend/app/research/__init__.py b/backend/app/research/__init__.py new file mode 100644 index 00000000..f751f44e --- /dev/null +++ b/backend/app/research/__init__.py @@ -0,0 +1,2 @@ +"""AI research layer for A-share opportunity discovery.""" + diff --git a/backend/app/research/catalyst_agent.py b/backend/app/research/catalyst_agent.py new file mode 100644 index 00000000..a6f5b201 --- /dev/null +++ b/backend/app/research/catalyst_agent.py @@ -0,0 +1,16 @@ +"""Catalyst summary agent.""" + +from __future__ import annotations + + +def build_catalyst_summary(theme_views: list[dict]) -> dict: + strong = [item for item in theme_views if item.get("heat_score", 0) >= 70] + themes = [item["theme"] for item in strong[:3]] + if themes: + summary = f"强催化/强热度主题集中在 {'、'.join(themes)}。" + elif theme_views: + summary = "主题热度存在但强催化不足，需等待新闻、政策或资金进一步确认。" + else: + summary = "暂未形成有效主题催化。" + return {"summary": summary, "strong_theme_count": len(strong), "themes": themes} + diff --git a/backend/app/research/feedback_agent.py b/backend/app/research/feedback_agent.py new file mode 100644 index 00000000..34ffe3d1 --- /dev/null +++ b/backend/app/research/feedback_agent.py @@ -0,0 +1,80 @@ +"""Review feedback weights for opportunity ranking.""" + +from __future__ import annotations + +from app.research.review_agent import build_research_review + + +async def build_ranking_feedback(days: int = 60) -> dict: + review = await build_research_review(days=days) + theme_weights = _weights_from_breakdown(review.get("theme_breakdown", []), positive=True) + chain_weights = _weights_from_breakdown(review.get("chain_breakdown", []), positive=True) + signal_weights = _weights_from_breakdown(review.get("signal_breakdown", []), positive=True) + risk_weights = _weights_from_breakdown(review.get("risk_breakdown", []), positive=False) + return { + "days": days, + "sample_count": review.get("sample_count", 0), + "tracked_count": review.get("tracked_count", 0), + "theme_weights": theme_weights, + "chain_weights": chain_weights, + "signal_weights": signal_weights, + "risk_weights": risk_weights, + "summary": review.get("summary", {}), + } + + +def apply_feedback_to_card(card: dict, rec, risk_alerts: list[dict], feedback: dict | None) -> dict: + if not feedback or int(feedback.get("tracked_count") or 0) < 5: + return {**card, "review_adjustment": 0, "adjusted_score": card.get("score", 0), "review_feedback": []} + + adjustments: list[dict] = [] + theme = str(card.get("theme") or "") + chain_node = str(card.get("chain_node") or "") + signal = str(getattr(rec, "entry_signal_type", "") or "") + + _append_adjustment(adjustments, "主题", theme, feedback.get("theme_weights", {}).get(theme, 0)) + _append_adjustment(adjustments, "环节", chain_node, feedback.get("chain_weights", {}).get(chain_node, 0)) + _append_adjustment(adjustments, "信号", signal, feedback.get("signal_weights", {}).get(signal, 0)) + + ts_code = str(card.get("ts_code") or "") + for risk in risk_alerts: + if risk.get("ts_code") and risk.get("ts_code") != ts_code: + continue + risk_type = str(risk.get("risk_type") or "") + _append_adjustment(adjustments, "风险", risk_type, feedback.get("risk_weights", {}).get(risk_type, 0)) + + total = round(sum(item["delta"] for item in adjustments), 1) + base_score = float(card.get("score") or 0) + adjusted = round(max(0, min(100, base_score + total)), 1) + return { + **card, + "review_adjustment": total, + "adjusted_score": adjusted, + "review_feedback": adjustments[:5], + } + + +def _weights_from_breakdown(items: list[dict], positive: bool) -> dict[str, float]: + weights: dict[str, float] = {} + for item in items: + label = str(item.get("label") or "") + tracked = int(item.get("tracked_count") or 0) + if not label or tracked < 2: + continue + avg_return = float(item.get("avg_return") or 0) + win_rate = float(item.get("win_rate") or 0) + if positive: + raw = (win_rate - 50) / 12 + avg_return * 0.65 + delta = max(-4.0, min(6.0, raw)) + else: + raw = avg_return * 0.85 - max(0, 45 - win_rate) / 10 + delta = max(-8.0, min(0.0, raw)) + if abs(delta) >= 0.5: + weights[label] = round(delta, 1) + return weights + + +def _append_adjustment(adjustments: list[dict], source: str, label: str, delta: float) -> None: + if not label or not delta: + return + adjustments.append({"source": source, "label": label, "delta": round(float(delta), 1)}) diff --git a/backend/app/research/industry_chain_agent.py b/backend/app/research/industry_chain_agent.py new file mode 100644 index 00000000..570edc34 --- /dev/null +++ b/backend/app/research/industry_chain_agent.py @@ -0,0 +1,245 @@ +"""Theme and industry-chain mapping. + +The first version is intentionally deterministic and editable. It gives the +research layer a stable vocabulary before LLM-assisted updates are introduced. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +from typing import Any + +from sqlalchemy import text + +from app.db import tables +from app.db.database import get_db + + +@dataclass(frozen=True) +class ThemeChain: + theme: str + aliases: tuple[str, ...] + nodes: tuple[str, ...] + logic: str + stage: str = "mid" + lifecycle_status: str = "观察期" + + +THEME_CHAINS: tuple[ThemeChain, ...] = ( + ThemeChain("AI算力", ("AI", "算力", "人工智能", "光模块", "服务器", "液冷", "PCB", "CPO"), ("光模块", "PCB", "服务器", "液冷", "IDC", "电源", "铜缆高速连接"), "AI 基础设施扩张带动上游硬件和数据中心链条。"), + ThemeChain("机器人", ("机器人", "人形机器人", "机器人装备", "减速器", "伺服", "控制器"), ("减速器", "伺服系统", "控制器", "传感器", "本体制造", "机器视觉"), "人形机器人产业趋势扩散，重点观察核心零部件和前排整机。"), + ThemeChain("低空经济", ("低空", "飞行汽车", "eVTOL", "通航", "无人机"), ("整机", "飞控", "电池", "空管", "材料", "运营服务"), "政策驱动低空基础设施和飞行器产业链加速。"), + ThemeChain("创新药", ("创新药", "医药", "CXO", "生物医药", "减肥药"), ("创新药企", "CXO", "原料药", "医疗器械", "商业化平台"), "政策、临床进展和出海订单共同驱动医药成长线。"), + ThemeChain("军工", ("军工", "航天", "航空", "卫星", "船舶"), ("航空装备", "航天电子", "卫星互联网", "船舶", "军工材料"), "订单周期、装备升级和地缘扰动共同驱动军工链。"), + ThemeChain("固态电池", ("固态电池", "锂电", "电池", "新能源车"), ("电解质", "正极材料", "负极材料", "隔膜", "设备", "整车验证"), "电池技术迭代推动材料和设备环节重估。"), + ThemeChain("半导体", ("半导体", "芯片", "集成电路", "存储", "先进封装"), ("设备", "材料", "设计", "制造", "封测", "存储"), "国产替代和周期复苏共同影响半导体链。"), + ThemeChain("传媒游戏", ("传媒", "游戏", "短剧", "影视", "AIGC"), ("游戏", "影视院线", "短剧", "营销", "版权IP", "AIGC应用"), "内容供给、AI 应用和监管周期变化驱动传媒弹性。"), + ThemeChain("化工材料", ("化工", "材料", "有机硅", "氟化工", "化纤"), ("基础化工", "新材料", "氟化工", "有机硅", "化纤", "电子化学品"), "价格周期和新材料需求共同影响化工材料链。"), + ThemeChain("有色金属", ("有色", "铜", "铝", "黄金", "稀土", "锂"), ("铜铝", "贵金属", "稀土", "锂资源", "加工材"), "通胀预期、供需缺口和新能源需求影响资源品。"), + ThemeChain("新能源", ("光伏", "风电", "储能", "新能源", "逆变器"), ("光伏组件", "逆变器", "储能", "风电设备", "电网设备"), "装机需求和价格出清影响新能源链条。"), + ThemeChain("消费电子", ("消费电子", "苹果", "MR", "折叠屏", "智能穿戴"), ("结构件", "面板", "光学", "声学", "组装", "芯片"), "终端创新周期带动零部件弹性。"), +) + + +def resolve_theme(name: str) -> ThemeChain | None: + normalized = name or "" + for theme in THEME_CHAINS: + if theme.theme in normalized or any(alias in normalized for alias in theme.aliases): + return theme + return None + + +async def ensure_theme_knowledge_seeded() -> None: + """Seed editable theme knowledge from the built-in v1 map when empty.""" + async with get_db() as db: + result = await db.execute(text("SELECT COUNT(*) AS count FROM theme_knowledge")) + count = int(result.fetchone()._mapping["count"]) + if count: + return + now_values = [] + for index, theme in enumerate(THEME_CHAINS): + now_values.append({ + "theme_name": theme.theme, + "aliases_json": json.dumps(list(theme.aliases), ensure_ascii=False), + "logic_summary": theme.logic, + "lifecycle_status": theme.lifecycle_status, + "stage": theme.stage, + "is_active": True, + "sort_order": index, + }) + await db.execute(tables.theme_knowledge_table.insert(), now_values) + + node_values = [] + for theme_index, theme in enumerate(THEME_CHAINS): + for node_index, node in enumerate(theme.nodes): + node_values.append({ + "theme_name": theme.theme, + "chain_node": node, + "related_stocks": "[]", + "leader_stocks": "[]", + "node_role": "", + "is_active": True, + "sort_order": theme_index * 100 + node_index, + }) + if node_values: + await db.execute(tables.theme_chain_knowledge_table.insert(), node_values) + await db.commit() + + +async def load_theme_chain_library() -> list[dict[str, Any]]: + await ensure_theme_knowledge_seeded() + async with get_db() as db: + themes_result = await db.execute( + text( + "SELECT * FROM theme_knowledge " + "WHERE is_active = 1 ORDER BY sort_order ASC, id ASC" + ) + ) + nodes_result = await db.execute( + text( + "SELECT * FROM theme_chain_knowledge " + "WHERE is_active = 1 ORDER BY sort_order ASC, id ASC" + ) + ) + node_map: dict[str, list[dict[str, Any]]] = {} + for row in nodes_result.fetchall(): + item = dict(row._mapping) + node_map.setdefault(str(item.get("theme_name") or ""), []).append(item) + + library = [] + for row in themes_result.fetchall(): + item = dict(row._mapping) + theme_name = str(item.get("theme_name") or "") + nodes = node_map.get(theme_name, []) + library.append({ + "theme": theme_name, + "aliases": _safe_json_list(item.get("aliases_json")), + "logic": item.get("logic_summary") or "", + "stage": item.get("stage") or "mid", + "lifecycle_status": item.get("lifecycle_status") or "观察期", + "chain_nodes": [node.get("chain_node") for node in nodes if node.get("chain_node")] or ["未归类"], + "chain_items": [ + { + "chain_node": node.get("chain_node") or "", + "related_stocks": _safe_json_list(node.get("related_stocks")), + "leader_stocks": _safe_json_list(node.get("leader_stocks")), + "node_role": node.get("node_role") or "", + } + for node in nodes + ], + }) + return library + + +def resolve_theme_from_library(name: str, library: list[dict[str, Any]]) -> dict[str, Any] | None: + normalized = name or "" + for theme in library: + theme_name = str(theme.get("theme") or "") + aliases = [str(alias) for alias in theme.get("aliases", [])] + if theme_name in normalized or any(alias and alias in normalized for alias in aliases): + return theme + return None + + +def map_sector_to_chain(sector_name: str, leading_stocks: list[dict[str, Any]] | None = None) -> dict[str, Any]: + theme = resolve_theme(sector_name) + if not theme: + return { + "theme": sector_name or "未归类", + "logic": "暂未命中内置产业链图谱，保留为未归类主题等待后续研究补全。", + "chain_nodes": ["未归类"], + "leader_stocks": leading_stocks or [], + } + return { + "theme": theme.theme, + "logic": theme.logic, + "chain_nodes": list(theme.nodes), + "chain_items": [ + {"chain_node": node, "related_stocks": [], "leader_stocks": [], "node_role": ""} + for node in theme.nodes + ], + "leader_stocks": leading_stocks or [], + } + + +def map_sector_to_chain_from_library( + sector_name: str, + leading_stocks: list[dict[str, Any]] | None, + library: list[dict[str, Any]], +) -> dict[str, Any]: + theme = resolve_theme_from_library(sector_name, library) + if not theme: + return map_sector_to_chain(sector_name, leading_stocks) + return { + "theme": theme["theme"], + "logic": theme["logic"] or "主题逻辑待补充。", + "chain_nodes": theme["chain_nodes"], + "chain_items": theme.get("chain_items", []), + "leader_stocks": leading_stocks or [], + "stage": theme.get("stage") or "mid", + "lifecycle_status": theme.get("lifecycle_status") or "观察期", + } + + +def infer_chain_node(theme_name: str, stock_name: str = "", sector_name: str = "") -> str: + theme = resolve_theme(theme_name) or resolve_theme(sector_name) + if not theme: + return "未归类" + text = f"{stock_name}{sector_name}{theme_name}" + for node in theme.nodes: + if node in text: + return node + return theme.nodes[0] if theme.nodes else "未归类" + + +def infer_chain_position_from_theme_view(theme_view: dict[str, Any], ts_code: str = "", stock_name: str = "") -> dict[str, str]: + """Infer a candidate's industry-chain node and role from editable theme knowledge.""" + chain_items = theme_view.get("chain_items") or [] + for item in chain_items: + node = str(item.get("chain_node") or "") + node_role = str(item.get("node_role") or "") + if _stock_in_list(ts_code, stock_name, item.get("leader_stocks") or []): + return {"chain_node": node or "未归类", "stock_role": node_role or "核心股"} + if _stock_in_list(ts_code, stock_name, item.get("related_stocks") or []): + return {"chain_node": node or "未归类", "stock_role": node_role or "相关股"} + + text_value = f"{stock_name}{theme_view.get('raw_sector', '')}{theme_view.get('theme', '')}" + for item in chain_items: + node = str(item.get("chain_node") or "") + if node and node in text_value: + return {"chain_node": node, "stock_role": str(item.get("node_role") or "环节标的")} + + nodes = theme_view.get("chain_nodes") or [] + fallback_node = str(nodes[0]) if nodes else infer_chain_node(str(theme_view.get("theme") or ""), stock_name, str(theme_view.get("raw_sector") or "")) + return {"chain_node": fallback_node or "未归类", "stock_role": "待归类"} + + +def _safe_json_list(value: Any) -> list[Any]: + if isinstance(value, list): + return value + if not value: + return [] + try: + parsed = json.loads(str(value)) + return parsed if isinstance(parsed, list) else [] + except Exception: + return [] + + +def _stock_in_list(ts_code: str, stock_name: str, values: list[Any]) -> bool: + code = (ts_code or "").upper() + name = stock_name or "" + for value in values: + if isinstance(value, dict): + raw_code = str(value.get("ts_code") or value.get("code") or "").upper() + raw_name = str(value.get("name") or "") + if raw_code and raw_code == code: + return True + if raw_name and raw_name == name: + return True + else: + text_value = str(value) + if text_value and (text_value.upper() == code or text_value == name): + return True + return False diff --git a/backend/app/research/market_agent.py b/backend/app/research/market_agent.py new file mode 100644 index 00000000..09f6acfb --- /dev/null +++ b/backend/app/research/market_agent.py @@ -0,0 +1,51 @@ +"""Market regime research agent.""" + +from __future__ import annotations + +from typing import Any + + +def build_market_view(market_temp: Any, strategy_profile: dict | None = None) -> dict: + temp = float(getattr(market_temp, "temperature", 0) or 0) + up = int(getattr(market_temp, "up_count", 0) or 0) + down = int(getattr(market_temp, "down_count", 0) or 0) + limit_up = int(getattr(market_temp, "limit_up_count", 0) or 0) + limit_down = int(getattr(market_temp, "limit_down_count", 0) or 0) + breadth_total = max(up + down, 1) + up_ratio = up / breadth_total + + if temp >= 70 and limit_up >= 60: + regime = "bullish_mainline" + summary = "市场情绪偏强，主线进攻窗口打开。" + elif temp >= 55 and up_ratio >= 0.52: + regime = "bullish_rotation" + summary = "市场偏强但仍需确认主线持续性，适合围绕前排轮动。" + elif temp >= 40: + regime = "range_rotation" + summary = "市场震荡轮动，优先等待回踩和承接确认。" + elif limit_down > max(limit_up, 1): + regime = "risk_off" + summary = "市场风险偏好较弱，优先防守和降低出手频率。" + else: + regime = "defensive_watch" + summary = "市场温度偏低，保留观察，不主动扩大仓位。" + + confidence = min(0.95, max(0.45, abs(temp - 50) / 60 + abs(up_ratio - 0.5) + 0.45)) + stance = (strategy_profile or {}).get("market_stance") or "" + if stance: + summary = f"{summary}{stance}策略生效。" + + return { + "regime": regime, + "confidence": round(confidence, 2), + "summary": summary, + "temperature": round(temp, 1), + "up_count": up, + "down_count": down, + "limit_up_count": limit_up, + "limit_down_count": limit_down, + "source": getattr(market_temp, "source", "snapshot"), + "data_status": getattr(market_temp, "data_status", "fresh"), + "source_detail": getattr(market_temp, "source_detail", ""), + "limit_counts_reliable": bool(getattr(market_temp, "limit_counts_reliable", False)), + } diff --git a/backend/app/research/ranking_agent.py b/backend/app/research/ranking_agent.py new file mode 100644 index 00000000..23938254 --- /dev/null +++ b/backend/app/research/ranking_agent.py @@ -0,0 +1,193 @@ +"""Opportunity card ranking agent.""" + +from __future__ import annotations + +from typing import Any + +from app.research.feedback_agent import apply_feedback_to_card +from app.research.industry_chain_agent import infer_chain_node + + +def build_opportunity_cards( + recommendations: list[Any], + stock_notes: list[dict], + risk_alerts: list[dict], + feedback: dict | None = None, +) -> list[dict]: + note_map = {item["ts_code"]: item for item in stock_notes} + rejected = {item["ts_code"] for item in risk_alerts if item.get("reject") and item.get("ts_code")} + risk_by_code: dict[str, list[dict]] = {} + global_risks: list[dict] = [] + for risk in risk_alerts: + if risk.get("ts_code"): + risk_by_code.setdefault(risk["ts_code"], []).append(risk) + else: + global_risks.append(risk) + cards: list[dict] = [] + for rec in recommendations: + ts_code = getattr(rec, "ts_code", "") + if ts_code in rejected: + continue + note = note_map.get(ts_code, {}) + action_plan = getattr(rec, "action_plan", "观察") or "观察" + if action_plan == "可操作": + opportunity_type = "可操作" + elif action_plan == "重点关注": + opportunity_type = "等确认" + else: + hint = ((getattr(rec, "decision_trace", {}) or {}).get("position_adjustment") or {}).get("hint", "") + opportunity_type = "等回踩" if hint == "wait_pullback" else "仅观察" + theme = getattr(rec, "sector", "") or note.get("theme") or "未归类" + alpha_profile = _build_alpha_profile(rec, note, risk_by_code.get(ts_code, []) + global_risks) + card = { + "ts_code": ts_code, + "name": getattr(rec, "name", ""), + "theme": theme, + "chain_node": note.get("chain_node") or infer_chain_node(theme, getattr(rec, "name", ""), theme), + "stock_role": note.get("stock_role", "待归类"), + "opportunity_type": opportunity_type, + "score": round(float(getattr(rec, "score", 0) or 0), 1), + "logic_score": note.get("logic_score", 0), + "action_plan": action_plan, + "trigger": getattr(rec, "trigger_condition", "") or getattr(rec, "entry_timing", ""), + "invalid_condition": note.get("invalid_condition") or getattr(rec, "invalidation_condition", "") or getattr(rec, "risk_note", ""), + "logic_summary": note.get("logic_summary", ""), + **alpha_profile, + } + cards.append(apply_feedback_to_card(card, rec, risk_by_code.get(ts_code, []) + global_risks, feedback)) + cards.sort( + key=lambda item: ( + {"可操作": 3, "等确认": 2, "等回踩": 1, "仅观察": 0}.get(item["opportunity_type"], 0), + item.get("alpha_score", 0), + item.get("ambush_score", 0), + item.get("adjusted_score", item["score"]), + item["logic_score"], + ), + reverse=True, + ) + return cards[:20] + + +def _build_alpha_profile(rec: Any, note: dict, risks: list[dict]) -> dict: + trace = getattr(rec, "decision_trace", {}) or {} + position = trace.get("position_adjustment") or {} + hint = position.get("hint", "") + action_plan = getattr(rec, "action_plan", "观察") or "观察" + score = float(getattr(rec, "score", 0) or 0) + market_score = float(getattr(rec, "market_temp_score", 0) or 0) + sector_score = float(getattr(rec, "sector_score", 0) or 0) + capital_score = float(getattr(rec, "capital_score", 0) or 0) + technical_score = float(getattr(rec, "technical_score", 0) or 0) + valuation_score = float(getattr(rec, "valuation_score", 0) or 0) + position_score = float(getattr(rec, "position_score", 0) or 0) + logic_score = float(note.get("logic_score", 0) or 0) + risk_penalty = min(sum(18 if item.get("reject") else 7 for item in risks), 35) + + beta_dependency_score = _clamp( + 20 + + market_score * 0.55 + + max(0, sector_score - capital_score) * 0.2 + - min(capital_score, 20) * 0.15 + ) + alpha_score = _clamp( + score * 0.25 + + sector_score * 0.18 + + capital_score * 0.22 + + technical_score * 0.16 + + logic_score * 0.14 + + valuation_score * 0.08 + - beta_dependency_score * 0.12 + - risk_penalty + ) + ambush_score = _clamp( + position_score * 0.26 + + valuation_score * 0.22 + + capital_score * 0.18 + + technical_score * 0.16 + + (12 if hint in {"wait_confirm", "actionable_pullback", "wait_pullback"} else 0) + + (8 if action_plan in {"重点关注", "观察"} else 0) + - risk_penalty * 0.7 + ) + expectation_gap_score = _clamp( + logic_score * 0.28 + + sector_score * 0.2 + + capital_score * 0.2 + + valuation_score * 0.16 + + (10 if note.get("generated_by") == "llm" else 4) + - market_score * 0.1 + - risk_penalty * 0.6 + ) + + risk_gate = _risk_gate(risks) + alpha_type = _alpha_type(action_plan, hint, ambush_score, expectation_gap_score, beta_dependency_score) + beta_dependency = _beta_label(beta_dependency_score) + setup_quality = _setup_quality(alpha_score, ambush_score, risk_gate) + + return { + "alpha_type": alpha_type, + "alpha_score": round(alpha_score, 1), + "beta_dependency": beta_dependency, + "beta_dependency_score": round(beta_dependency_score, 1), + "ambush_score": round(ambush_score, 1), + "expectation_gap_score": round(expectation_gap_score, 1), + "risk_gate": risk_gate, + "setup_quality": setup_quality, + "alpha_reason": _alpha_reason(alpha_type, beta_dependency, ambush_score, expectation_gap_score, risk_gate), + } + + +def _alpha_type(action_plan: str, hint: str, ambush_score: float, expectation_gap_score: float, beta_dependency_score: float) -> str: + if ambush_score >= 70 and action_plan != "可操作": + return "低位埋伏" + if hint == "wait_pullback": + return "强势等回踩" + if hint == "wait_confirm" or action_plan == "重点关注": + return "等确认" + if expectation_gap_score >= 72 and beta_dependency_score <= 65: + return "预期差机会" + if action_plan == "可操作": + return "趋势确认" + return "观察线索" + + +def _risk_gate(risks: list[dict]) -> str: + if any(item.get("reject") for item in risks): + return "否决" + if risks: + return "预警" + return "通过" + + +def _beta_label(score: float) -> str: + if score >= 70: + return "高" + if score >= 45: + return "中" + return "低" + + +def _setup_quality(alpha_score: float, ambush_score: float, risk_gate: str) -> str: + if risk_gate == "否决": + return "不参与" + if alpha_score >= 78 and ambush_score >= 65: + return "优先研究" + if alpha_score >= 65 or ambush_score >= 70: + return "可跟踪" + return "仅观察" + + +def _alpha_reason(alpha_type: str, beta_dependency: str, ambush_score: float, expectation_gap_score: float, risk_gate: str) -> str: + if risk_gate == "否决": + return "风险门槛未通过，先排除。" + parts = [f"{alpha_type}，Beta依赖{beta_dependency}"] + if ambush_score >= 70: + parts.append("位置更适合提前跟踪") + if expectation_gap_score >= 70: + parts.append("存在预期差") + if risk_gate == "预警": + parts.append("但有风险预警") + return "；".join(parts) + "。" + + +def _clamp(value: float, low: float = 0, high: float = 100) -> float: + return max(low, min(high, value)) diff --git a/backend/app/research/report_agent.py b/backend/app/research/report_agent.py new file mode 100644 index 00000000..a8e2f65f --- /dev/null +++ b/backend/app/research/report_agent.py @@ -0,0 +1,354 @@ +"""Build and persist daily AI research reports.""" + +from __future__ import annotations + +import json +from datetime import datetime +from typing import Any + +from sqlalchemy import text + +from app.db.database import get_db +from app.db import tables +from app.research.catalyst_agent import build_catalyst_summary +from app.research.feedback_agent import build_ranking_feedback +from app.research.market_agent import build_market_view +from app.research.ranking_agent import build_opportunity_cards +from app.research.risk_agent import build_risk_alerts +from app.research.sector_agent import build_theme_views, build_theme_views_async +from app.research.stock_research_agent import build_stock_research_notes, build_stock_research_notes_sync + + +def _latest_scan_from_result(result: dict, scan_session: str) -> dict: + for item in result.get("scan_logs", []) or []: + if item.get("stage") == "final_filter": + return item + return { + "scan_session": scan_session, + "scan_mode": result.get("scan_mode", ""), + "status": "empty" if not result.get("recommendations") else "ok", + "summary": "", + "elimination_reasons": {}, + "created_at": datetime.now().isoformat(), + } + + +def build_research_report(result: dict, scan_session: str) -> dict: + """Build a report with deterministic stock notes. + + This remains available for fast API fallbacks and tests. Normal scans use + build_research_report_async so the stock research notes can call the LLM. + """ + return _assemble_research_report( + result, + scan_session, + build_stock_research_notes_sync, + ) + + +async def build_research_report_async(result: dict, scan_session: str) -> dict: + return await _assemble_research_report_async(result, scan_session) + + +def _common_inputs(result: dict, scan_session: str) -> tuple[Any, dict, list, list, dict, dict, list, dict, list]: + market_temp = result.get("market_temp") + strategy_profile = result.get("strategy_profile") or {} + sectors = result.get("hot_sectors", []) or [] + recommendations = result.get("recommendations", []) or [] + latest_scan = result.get("latest_scan") or _latest_scan_from_result(result, scan_session) + + theme_views = build_theme_views(sectors) + _calibrate_market_from_themes(market_temp, theme_views) + market_view = build_market_view(market_temp, strategy_profile) + catalyst = build_catalyst_summary(theme_views) + risks = build_risk_alerts(recommendations, market_view, latest_scan) + return market_temp, strategy_profile, sectors, recommendations, latest_scan, market_view, theme_views, catalyst, risks + + +def _assemble_research_report(result: dict, scan_session: str, notes_builder) -> dict: + market_temp, _, _, recommendations, latest_scan, market_view, theme_views, catalyst, risks = _common_inputs(result, scan_session) + stock_notes = notes_builder(recommendations, theme_views) + return _finalize_research_report(result, scan_session, market_temp, recommendations, latest_scan, market_view, theme_views, catalyst, risks, stock_notes, None, None) + + +async def _assemble_research_report_async(result: dict, scan_session: str) -> dict: + market_temp = result.get("market_temp") + strategy_profile = result.get("strategy_profile") or {} + sectors = result.get("hot_sectors", []) or [] + recommendations = result.get("recommendations", []) or [] + latest_scan = result.get("latest_scan") or _latest_scan_from_result(result, scan_session) + + theme_views = await build_theme_views_async(sectors) + _calibrate_market_from_themes(market_temp, theme_views) + market_view = build_market_view(market_temp, strategy_profile) + catalyst = build_catalyst_summary(theme_views) + risks = build_risk_alerts(recommendations, market_view, latest_scan) + stock_notes = await build_stock_research_notes(recommendations, theme_views, risks) + feedback = await build_ranking_feedback(days=60) + data_quality = await _build_data_quality_report(market_view) + return _finalize_research_report(result, scan_session, market_temp, recommendations, latest_scan, market_view, theme_views, catalyst, risks, stock_notes, feedback, data_quality) + + +def _finalize_research_report( + result: dict, + scan_session: str, + market_temp: Any, + recommendations: list, + latest_scan: dict, + market_view: dict, + theme_views: list[dict], + catalyst: dict, + risks: list[dict], + stock_notes: list[dict], + feedback: dict | None, + data_quality: dict | None, +) -> dict: + opportunities = build_opportunity_cards(recommendations, stock_notes, risks, feedback) + trade_date = getattr(market_temp, "trade_date", "") or datetime.now().strftime("%Y%m%d") + + if opportunities: + no_trade_reason = {"has_scan": True, "reason": "", "blocked_by": []} + else: + elimination = latest_scan.get("elimination_reasons") or {} + if elimination: + reason = "；".join(f"{k} {v}只" for k, v in list(elimination.items())[:3]) + elif recommendations: + reason = "候选存在，但机会卡被风险或动作分层过滤。" + else: + reason = "本轮扫描没有形成满足条件的交易候选。" + no_trade_reason = {"has_scan": True, "reason": reason, "blocked_by": list(elimination.keys())[:5]} + + top_theme_names = [item["theme"] for item in theme_views[:3]] + report = { + "trade_date": trade_date, + "scan_session": scan_session, + "scan_mode": result.get("scan_mode", ""), + "scanned_at": datetime.now().isoformat(), + "market_view": market_view, + "theme_views": theme_views, + "industry_chain_map": [ + { + "theme": item["theme"], + "chain_nodes": item["chain_nodes"], + "chain_items": item.get("chain_items", []), + "leader_stocks": item.get("leader_stocks", []), + } + for item in theme_views + ], + "catalyst": catalyst, + "stock_research_notes": stock_notes, + "opportunity_cards": opportunities, + "risk_alerts": risks, + "ranking_feedback": feedback or {}, + "risk_summary": { + "reject_count": sum(1 for item in risks if item.get("reject")), + "warning_count": sum(1 for item in risks if not item.get("reject")), + "types": sorted({item.get("risk_type", "") for item in risks if item.get("risk_type")}), + }, + "data_quality": data_quality or _fallback_data_quality(market_view), + "no_trade_reason": no_trade_reason, + "summary": { + "market": market_view["summary"], + "theme": f"当前主线关注 {'、'.join(top_theme_names)}。" if top_theme_names else "暂未形成清晰主线。", + "opportunity_count": len(opportunities), + "risk_count": len(risks), + }, + } + return report + + +async def _build_data_quality_report(market_view: dict) -> dict: + watched_sources = ("eastmoney", "tencent", "market_breadth") + since = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) + issues = [] + async with get_db() as db: + result = await db.execute( + text( + "SELECT source, level, message, created_at FROM error_logs " + "WHERE created_at >= :since AND (" + "source LIKE '%eastmoney%' OR source LIKE '%tencent%' OR source = 'market_breadth'" + ") ORDER BY created_at DESC, id DESC LIMIT 20" + ), + {"since": since.strftime("%Y-%m-%d %H:%M:%S")}, + ) + for row in result.fetchall(): + item = dict(row._mapping) + issues.append({ + "source": item.get("source", ""), + "level": item.get("level", ""), + "message": item.get("message", ""), + "created_at": str(item.get("created_at") or ""), + }) + + warnings = [] + market_status = market_view.get("data_status") or "fresh" + if market_status == "estimated": + warnings.append("全市场涨跌停使用实时行情阈值估算，非涨跌停池精确口径。") + if issues: + warnings.append("今日存在行情源失败记录，盘中结论已降级处理。") + + status = "degraded" if warnings or issues else "ok" + return { + "status": status, + "market_data_status": market_status, + "market_source": market_view.get("source", ""), + "limit_counts_reliable": bool(market_view.get("limit_counts_reliable", False)), + "warnings": warnings, + "issues": issues[:8], + } + + +def _fallback_data_quality(market_view: dict) -> dict: + market_status = market_view.get("data_status") or "fresh" + warnings = [] + if market_status == "estimated": + warnings.append("全市场涨跌停使用实时行情阈值估算。") + return { + "status": "degraded" if warnings else "ok", + "market_data_status": market_status, + "market_source": market_view.get("source", ""), + "limit_counts_reliable": bool(market_view.get("limit_counts_reliable", False)), + "warnings": warnings, + "issues": [], + } + + +def _calibrate_market_from_themes(market_temp: Any, theme_views: list[dict]) -> None: + if not market_temp or not theme_views: + return + theme_limit_up = sum(max(int(item.get("limit_up_count") or 0), 0) for item in theme_views) + if theme_limit_up <= int(getattr(market_temp, "limit_up_count", 0) or 0): + return + original_limit = int(getattr(market_temp, "limit_up_count", 0) or 0) + original_temp = float(getattr(market_temp, "temperature", 0) or 0) + market_temp.limit_up_count = theme_limit_up + if original_limit == 0: + market_temp.temperature = round(min(original_temp + min(theme_limit_up / 5, 8), 100), 1) + if not getattr(market_temp, "limit_counts_reliable", False): + market_temp.data_status = "estimated" + detail = getattr(market_temp, "source_detail", "") or "" + market_temp.source_detail = f"{detail};theme_limit_lower_bound" + + +async def save_research_report(report: dict) -> None: + trade_date = str(report.get("trade_date") or "") + scan_session = str(report.get("scan_session") or "manual") + now = datetime.now() + async with get_db() as db: + await db.execute(text("DELETE FROM research_reports WHERE scan_session = :session"), {"session": scan_session}) + await db.execute(text("DELETE FROM theme_maps WHERE scan_session = :session"), {"session": scan_session}) + await db.execute(text("DELETE FROM theme_chain_nodes WHERE scan_session = :session"), {"session": scan_session}) + await db.execute(text("DELETE FROM stock_research_notes WHERE scan_session = :session"), {"session": scan_session}) + await db.execute(text("DELETE FROM risk_events WHERE scan_session = :session"), {"session": scan_session}) + await db.execute(text("DELETE FROM opportunity_cards WHERE scan_session = :session"), {"session": scan_session}) + + await db.execute(tables.research_reports_table.insert().values( + scan_session=scan_session, + trade_date=trade_date, + market_summary=report.get("summary", {}).get("market", ""), + theme_summary=report.get("summary", {}).get("theme", ""), + no_trade_reason=json.dumps(report.get("no_trade_reason", {}), ensure_ascii=False), + report_json=json.dumps(report, ensure_ascii=False, default=str), + created_at=now, + )) + + for theme in report.get("theme_views", []): + await db.execute(tables.theme_maps_table.insert().values( + scan_session=scan_session, + trade_date=trade_date, + theme_name=theme.get("theme", ""), + stage=theme.get("stage", ""), + heat_score=theme.get("heat_score", 0), + logic_summary=theme.get("logic", ""), + lifecycle_status=theme.get("lifecycle_status", ""), + created_at=now, + )) + for node in theme.get("chain_nodes", []) or ["未归类"]: + chain_item = _chain_item_for_node(theme, node) + await db.execute(tables.theme_chain_nodes_table.insert().values( + scan_session=scan_session, + trade_date=trade_date, + theme_name=theme.get("theme", ""), + chain_node=node, + related_stocks=json.dumps(chain_item.get("related_stocks", []), ensure_ascii=False, default=str), + leader_stocks=json.dumps(chain_item.get("leader_stocks", []) or theme.get("leader_stocks", []), ensure_ascii=False, default=str), + created_at=now, + )) + + for note in report.get("stock_research_notes", []): + await db.execute(tables.stock_research_notes_table.insert().values( + scan_session=scan_session, + trade_date=trade_date, + ts_code=note.get("ts_code", ""), + name=note.get("name", ""), + theme=note.get("theme", "未归类"), + chain_node=note.get("chain_node", "未归类"), + logic_score=note.get("logic_score", 0), + logic_summary=note.get("logic_summary", ""), + evidence_json=json.dumps(note.get("evidence", []), ensure_ascii=False, default=str), + uncertainty=note.get("uncertainty", ""), + stock_role=note.get("stock_role", "待归类"), + disagreement=note.get("disagreement", ""), + invalid_condition=note.get("invalid_condition", ""), + generated_by=note.get("generated_by", "rules"), + created_at=now, + )) + + for risk in report.get("risk_alerts", []): + await db.execute(tables.risk_events_table.insert().values( + scan_session=scan_session, + trade_date=trade_date, + ts_code=risk.get("ts_code", ""), + risk_type=risk.get("risk_type", ""), + severity=risk.get("severity", "warning"), + reject=bool(risk.get("reject")), + reason=risk.get("reason", ""), + source=risk.get("source", "research_agent"), + created_at=now, + )) + + for card in report.get("opportunity_cards", []): + await db.execute(tables.opportunity_cards_table.insert().values( + scan_session=scan_session, + trade_date=trade_date, + ts_code=card.get("ts_code", ""), + name=card.get("name", ""), + theme=card.get("theme", ""), + chain_node=card.get("chain_node", "未归类"), + stock_role=card.get("stock_role", "待归类"), + opportunity_type=card.get("opportunity_type", "观察"), + score=card.get("adjusted_score", card.get("score", 0)), + alpha_type=card.get("alpha_type", "观察线索"), + alpha_score=card.get("alpha_score", 0), + beta_dependency=card.get("beta_dependency", "中"), + beta_dependency_score=card.get("beta_dependency_score", 0), + ambush_score=card.get("ambush_score", 0), + expectation_gap_score=card.get("expectation_gap_score", 0), + risk_gate=card.get("risk_gate", "通过"), + setup_quality=card.get("setup_quality", "仅观察"), + alpha_reason=card.get("alpha_reason", ""), + action_plan=card.get("action_plan", "观察"), + trigger=card.get("trigger", ""), + invalid_condition=card.get("invalid_condition", ""), + created_at=now, + )) + await db.commit() + + +def _chain_item_for_node(theme: dict, node: str) -> dict: + for item in theme.get("chain_items", []) or []: + if item.get("chain_node") == node: + return item + return {} + + +async def load_latest_research_report() -> dict | None: + async with get_db() as db: + result = await db.execute(text("SELECT report_json FROM research_reports ORDER BY created_at DESC, id DESC LIMIT 1")) + row = result.fetchone() + if not row: + return None + try: + return json.loads(row._mapping["report_json"] or "{}") + except Exception: + return None diff --git a/backend/app/research/review_agent.py b/backend/app/research/review_agent.py new file mode 100644 index 00000000..88efae84 --- /dev/null +++ b/backend/app/research/review_agent.py @@ -0,0 +1,187 @@ +"""Research review aggregation. + +This module connects recommendation tracking with the new research layer, so +we can review which themes, chain nodes, signals and risks are working. +""" + +from __future__ import annotations + +from collections import defaultdict +from datetime import datetime, timedelta +from typing import Any + +from sqlalchemy import text + +from app.db.database import get_db + + +async def build_research_review(days: int = 60) -> dict: + start = (datetime.now() - timedelta(days=max(1, days))).strftime("%Y-%m-%d") + rows = await _load_review_rows(start) + theme_rows = _breakdown(rows, "theme") + chain_rows = _breakdown(rows, "chain_node") + signal_rows = _breakdown(rows, "entry_signal_type") + risk_rows = _risk_breakdown(rows) + summary = _summary(rows, theme_rows, chain_rows, signal_rows, risk_rows) + return { + "days": days, + "sample_count": len(rows), + "tracked_count": sum(1 for item in rows if item.get("pct_from_entry") is not None), + "theme_breakdown": theme_rows, + "chain_breakdown": chain_rows, + "signal_breakdown": signal_rows, + "risk_breakdown": risk_rows, + "summary": summary, + } + + +async def _load_review_rows(start: str) -> list[dict[str, Any]]: + async with get_db() as db: + result = await db.execute( + text( + "WITH latest_tracking AS (" + " SELECT t.* FROM recommendation_tracking t " + " INNER JOIN (" + " SELECT recommendation_id, MAX(id) AS max_id " + " FROM recommendation_tracking GROUP BY recommendation_id" + " ) lt ON t.id = lt.max_id" + "), latest_notes AS (" + " SELECT n.* FROM stock_research_notes n " + " INNER JOIN (" + " SELECT ts_code, MAX(id) AS max_id " + " FROM stock_research_notes GROUP BY ts_code" + " ) ln ON n.id = ln.max_id" + "), risk_summary AS (" + " SELECT ts_code, " + " GROUP_CONCAT(DISTINCT risk_type) AS risk_types, " + " MAX(CASE WHEN reject = 1 THEN 1 ELSE 0 END) AS rejected " + " FROM risk_events GROUP BY ts_code" + ") " + "SELECT r.id, r.ts_code, r.name, r.sector, r.entry_signal_type, r.action_plan, " + " r.lifecycle_status, r.score, r.created_at, " + " lt.pct_from_entry, lt.max_return_pct, lt.max_drawdown_pct, lt.hit_target, " + " lt.hit_stop_loss, lt.close_reason, lt.review_note, " + " COALESCE(n.theme, r.sector, '未归类') AS theme, " + " COALESCE(n.chain_node, '未归类') AS chain_node, " + " COALESCE(n.stock_role, '待归类') AS stock_role, " + " COALESCE(n.logic_score, 0) AS logic_score, " + " COALESCE(rs.risk_types, '') AS risk_types, " + " COALESCE(rs.rejected, 0) AS risk_rejected " + "FROM recommendations r " + "LEFT JOIN latest_tracking lt ON lt.recommendation_id = r.id " + "LEFT JOIN latest_notes n ON n.ts_code = r.ts_code " + "LEFT JOIN risk_summary rs ON rs.ts_code = r.ts_code " + "WHERE r.created_at >= :start " + "ORDER BY r.created_at DESC, r.score DESC" + ), + {"start": start}, + ) + return [dict(row._mapping) for row in result.fetchall()] + + +def _breakdown(rows: list[dict[str, Any]], key: str) -> list[dict]: + groups: dict[str, list[dict]] = defaultdict(list) + for row in rows: + label = str(row.get(key) or "未归类") + groups[label].append(row) + return [_build_group(label, items) for label, items in groups.items() if label][:12] + + +def _risk_breakdown(rows: list[dict[str, Any]]) -> list[dict]: + groups: dict[str, list[dict]] = defaultdict(list) + for row in rows: + risk_types = [item for item in str(row.get("risk_types") or "").split(",") if item] + for risk_type in risk_types: + groups[risk_type].append(row) + result = [_build_group(label, items) for label, items in groups.items()] + return sorted(result, key=lambda item: (item["sample_count"], abs(item["avg_return"])), reverse=True)[:10] + + +def _build_group(label: str, items: list[dict]) -> dict: + tracked = [item for item in items if item.get("pct_from_entry") is not None] + wins = [item for item in tracked if float(item.get("pct_from_entry") or 0) > 0] + avg_return = _avg(tracked, "pct_from_entry") + avg_max_return = _avg(tracked, "max_return_pct") + avg_drawdown = _avg(tracked, "max_drawdown_pct") + hit_target = sum(1 for item in tracked if item.get("hit_target")) + hit_stop = sum(1 for item in tracked if item.get("hit_stop_loss")) + return { + "label": label, + "sample_count": len(items), + "tracked_count": len(tracked), + "win_rate": round(len(wins) / len(tracked) * 100, 1) if tracked else 0, + "avg_return": round(avg_return, 2), + "avg_max_return": round(avg_max_return, 2), + "avg_drawdown": round(avg_drawdown, 2), + "hit_target_count": hit_target, + "hit_stop_count": hit_stop, + "effectiveness": _effectiveness(len(tracked), len(wins), avg_return, hit_stop), + "top_samples": [ + { + "ts_code": item.get("ts_code"), + "name": item.get("name"), + "pct_from_entry": item.get("pct_from_entry"), + "max_return_pct": item.get("max_return_pct"), + "created_at": str(item.get("created_at") or "")[:10], + } + for item in sorted(tracked, key=lambda row: float(row.get("pct_from_entry") or 0), reverse=True)[:3] + ], + } + + +def _summary( + rows: list[dict], + themes: list[dict], + chains: list[dict], + signals: list[dict], + risks: list[dict], +) -> dict: + tracked = [item for item in rows if item.get("pct_from_entry") is not None] + strongest_theme = _first_by_effectiveness(themes) + strongest_chain = _first_by_effectiveness(chains) + strongest_signal = _first_by_effectiveness(signals) + weakest_risk = sorted(risks, key=lambda item: item["avg_return"])[:1] + headline = "等待形成研究复盘样本" + if tracked: + headline = f"近 {len(tracked)} 个跟踪样本，{strongest_theme.get('label', '主题')} 相对更有效" + return { + "headline": headline, + "strongest_theme": strongest_theme, + "strongest_chain": strongest_chain, + "strongest_signal": strongest_signal, + "weakest_risk": weakest_risk[0] if weakest_risk else {}, + "suggestions": _suggestions(strongest_theme, strongest_chain, strongest_signal, weakest_risk[0] if weakest_risk else {}), + } + + +def _first_by_effectiveness(items: list[dict]) -> dict: + eligible = [item for item in items if item["tracked_count"] >= 1] + if not eligible: + return {} + return sorted(eligible, key=lambda item: (item["effectiveness"], item["win_rate"], item["avg_return"]), reverse=True)[0] + + +def _suggestions(theme: dict, chain: dict, signal: dict, risk: dict) -> list[str]: + suggestions = [] + if theme: + suggestions.append(f"提高 {theme['label']} 方向的复盘权重，当前胜率 {theme['win_rate']}%。") + if chain: + suggestions.append(f"优先跟踪 {chain['label']} 环节，平均收益 {chain['avg_return']}%。") + if signal: + suggestions.append(f"信号上关注 {signal['label']}，减少低效入口。") + if risk and risk.get("avg_return", 0) < 0: + suggestions.append(f"命中 {risk['label']} 风险的样本表现偏弱，后续应降低排序或直接观察。") + return suggestions[:4] or ["样本不足，继续累积主题、环节和信号复盘数据。"] + + +def _effectiveness(tracked_count: int, wins: int, avg_return: float, hit_stop: int) -> float: + if tracked_count <= 0: + return 0 + win_rate = wins / tracked_count * 100 + stop_penalty = hit_stop / tracked_count * 18 + return round(win_rate * 0.55 + avg_return * 2.5 - stop_penalty + min(tracked_count, 8), 2) + + +def _avg(rows: list[dict], key: str) -> float: + values = [float(item.get(key) or 0) for item in rows if item.get(key) is not None] + return sum(values) / len(values) if values else 0 diff --git a/backend/app/research/risk_agent.py b/backend/app/research/risk_agent.py new file mode 100644 index 00000000..7d4d39e6 --- /dev/null +++ b/backend/app/research/risk_agent.py @@ -0,0 +1,334 @@ +"""Risk research agent. + +This v1 uses deterministic risk signals already present in the scan result. +External reduction/unlock/regulatory data can be added behind this interface. +""" + +from __future__ import annotations + +import logging +from datetime import datetime, timedelta +from typing import Any + +from app.config import settings +from app.data.tushare_client import tushare_client + +logger = logging.getLogger(__name__) + + +def build_risk_alerts(recommendations: list[Any], market_view: dict, latest_scan: dict | None = None) -> list[dict]: + alerts: list[dict] = [] + if market_view.get("regime") in {"risk_off", "defensive_watch"}: + alerts.append({ + "ts_code": "", + "risk_type": "market_regime", + "severity": "reject" if market_view.get("regime") == "risk_off" else "warning", + "reject": market_view.get("regime") == "risk_off", + "reason": market_view.get("summary", "市场环境偏弱"), + "source": "market_agent", + }) + + elimination = (latest_scan or {}).get("elimination_reasons") or {} + if elimination: + top_reason = sorted(elimination.items(), key=lambda item: item[1], reverse=True)[0] + alerts.append({ + "ts_code": "", + "risk_type": "filter_pressure", + "severity": "warning", + "reject": False, + "reason": f"最终过滤压力较高：{top_reason[0]} {top_reason[1]} 只", + "source": "scan_process_logs", + }) + + for rec in recommendations[:20]: + trace = getattr(rec, "decision_trace", {}) or {} + position_hint = (trace.get("position_adjustment") or {}).get("hint", "") + if position_hint in {"wait_pullback", "wait_confirm"}: + alerts.append({ + "ts_code": getattr(rec, "ts_code", ""), + "risk_type": "position_risk", + "severity": "warning", + "reject": False, + "reason": "位置或买点需要等待确认，避免追高。", + "source": "risk_agent", + }) + + if settings.research_risk_enabled: + alerts.extend(_build_external_stock_risks(recommendations[: settings.research_risk_stock_limit])) + + return _dedupe_alerts(alerts)[:40] + + +def _build_external_stock_risks(recommendations: list[Any]) -> list[dict]: + today = datetime.now() + today_s = today.strftime("%Y%m%d") + unlock_end = (today + timedelta(days=settings.risk_unlock_lookahead_days)).strftime("%Y%m%d") + holder_start = (today - timedelta(days=settings.risk_holder_trade_lookback_days)).strftime("%Y%m%d") + forecast_start = (today - timedelta(days=settings.risk_forecast_lookback_days)).strftime("%Y%m%d") + announcement_start = (today - timedelta(days=settings.risk_announcement_lookback_days)).strftime("%Y%m%d") + financial_start = (today - timedelta(days=settings.risk_financial_lookback_days)).strftime("%Y%m%d") + alerts: list[dict] = [] + + for rec in recommendations: + ts_code = getattr(rec, "ts_code", "") + name = getattr(rec, "name", "") or ts_code + if not ts_code: + continue + try: + alerts.extend(_unlock_risks(ts_code, name, today_s, unlock_end)) + alerts.extend(_holder_trade_risks(ts_code, name, holder_start, today_s)) + alerts.extend(_forecast_risks(ts_code, name, forecast_start, today_s)) + alerts.extend(_pledge_risks(ts_code, name)) + alerts.extend(_announcement_risks(ts_code, name, announcement_start, today_s)) + alerts.extend(_audit_risks(ts_code, name, financial_start, today_s)) + alerts.extend(_financial_statement_risks(ts_code, name, financial_start, today_s)) + except Exception as exc: + logger.debug("外部风险检查失败 %s: %s", ts_code, exc) + alerts.append({ + "ts_code": ts_code, + "risk_type": "risk_data_source", + "severity": "warning", + "reject": False, + "reason": f"{name} 外部风险数据读取失败，保留扫描内生风险判断。", + "source": "risk_agent", + }) + return alerts + + +_ANNOUNCEMENT_RISK_RULES: tuple[tuple[str, tuple[str, ...], str, bool], ...] = ( + ("regulatory", ("监管函", "问询函", "关注函", "警示函", "责令改正"), "warning", False), + ("penalty", ("处罚", "行政处罚", "纪律处分", "公开谴责", "通报批评"), "reject", True), + ("investigation", ("立案", "调查通知书", "涉嫌违法", "涉嫌信息披露违法"), "reject", True), + ("litigation", ("重大诉讼", "重大仲裁", "诉讼进展", "仲裁进展"), "warning", False), + ("asset_freeze", ("冻结", "轮候冻结", "司法冻结", "质押违约"), "reject", True), + ("accounting_risk", ("会计差错", "前期差错", "非标准审计", "保留意见", "无法表示意见", "否定意见"), "reject", True), +) + + +def _unlock_risks(ts_code: str, name: str, start: str, end: str) -> list[dict]: + df = tushare_client.get_share_float(ts_code, start, end) + if df.empty: + return [] + alerts = [] + for _, row in df.head(3).iterrows(): + ratio = _float(row.get("float_ratio")) + shares = _float(row.get("float_share")) + date = str(row.get("float_date") or "") + holder = str(row.get("holder_name") or "限售股东") + if ratio >= settings.risk_unlock_reject_ratio: + severity, reject = "reject", True + elif ratio >= 5 or shares >= 5000: + severity, reject = "warning", False + else: + continue + alerts.append({ + "ts_code": ts_code, + "risk_type": "unlock", + "severity": severity, + "reject": reject, + "reason": f"{name} {date} 存在解禁，比例约{ratio:g}%（{holder}），需规避供给冲击。", + "source": "tushare.share_float", + }) + return alerts + + +def _holder_trade_risks(ts_code: str, name: str, start: str, end: str) -> list[dict]: + df = tushare_client.get_holder_trade(ts_code, start, end) + if df.empty: + return [] + alerts = [] + for _, row in df.head(5).iterrows(): + direction = str(row.get("in_de") or "") + ratio = abs(_float(row.get("change_ratio"))) + volume = abs(_float(row.get("change_vol"))) + ann_date = str(row.get("ann_date") or "") + holder = str(row.get("holder_name") or "股东") + is_reduce = "减" in direction or direction.upper() in {"D", "DECREASE"} + if not is_reduce: + continue + if ratio >= 2 or volume >= 3000: + severity, reject = "reject", ratio >= 5 + alerts.append({ + "ts_code": ts_code, + "risk_type": "holder_reduce", + "severity": severity, + "reject": reject, + "reason": f"{name} {ann_date} 披露股东减持：{holder}，变动比例约{ratio:g}%。", + "source": "tushare.stk_holdertrade", + }) + return alerts + + +def _forecast_risks(ts_code: str, name: str, start: str, end: str) -> list[dict]: + df = tushare_client.get_forecast(ts_code, start, end) + if df.empty: + return [] + alerts = [] + negative_keywords = ("预减", "亏损", "首亏", "续亏", "略减", "不确定") + for _, row in df.head(3).iterrows(): + forecast_type = str(row.get("type") or "") + summary = str(row.get("summary") or row.get("change_reason") or "") + ann_date = str(row.get("ann_date") or "") + min_change = _float(row.get("p_change_min")) + max_change = _float(row.get("p_change_max")) + negative = any(word in forecast_type or word in summary for word in negative_keywords) or max_change < -20 + if not negative: + continue + reject = forecast_type in {"首亏", "续亏"} or max_change < -50 + alerts.append({ + "ts_code": ts_code, + "risk_type": "earnings_forecast", + "severity": "reject" if reject else "warning", + "reject": reject, + "reason": f"{name} {ann_date} 业绩预告偏负面：{forecast_type}，变动区间约{min_change:g}%~{max_change:g}%。", + "source": "tushare.forecast", + }) + return alerts + + +def _pledge_risks(ts_code: str, name: str) -> list[dict]: + df = tushare_client.get_pledge_stat(ts_code) + if df.empty: + return [] + row = df.iloc[-1] + ratio = _float(row.get("pledge_ratio")) + if ratio < 30: + return [] + reject = ratio >= settings.risk_pledge_reject_ratio + return [{ + "ts_code": ts_code, + "risk_type": "pledge", + "severity": "reject" if reject else "warning", + "reject": reject, + "reason": f"{name} 股权质押比例约{ratio:g}%，需关注平仓和融资风险。", + "source": "tushare.pledge_stat", + }] + + +def _announcement_risks(ts_code: str, name: str, start: str, end: str) -> list[dict]: + df = tushare_client.get_announcements(ts_code, start, end) + if df.empty: + return [] + alerts = [] + for _, row in df.head(80).iterrows(): + title = str(row.get("title") or "") + ann_date = str(row.get("ann_date") or "") + if not title: + continue + for risk_type, keywords, severity, reject in _ANNOUNCEMENT_RISK_RULES: + if any(keyword in title for keyword in keywords): + alerts.append({ + "ts_code": ts_code, + "risk_type": risk_type, + "severity": severity, + "reject": reject, + "reason": f"{name} {ann_date} 公告命中风险：{title[:90]}", + "source": "tushare.anns_d", + }) + break + return alerts[:6] + + +def _audit_risks(ts_code: str, name: str, start: str, end: str) -> list[dict]: + df = tushare_client.get_fina_audit(ts_code, start, end) + if df.empty: + return [] + alerts = [] + for _, row in df.tail(3).iterrows(): + result = str(row.get("audit_result") or "") + ann_date = str(row.get("ann_date") or "") + agency = str(row.get("audit_agency") or "审计机构") + severity, reject = _classify_audit_opinion(result) + if not severity: + continue + alerts.append({ + "ts_code": ts_code, + "risk_type": "audit_opinion", + "severity": severity, + "reject": reject, + "reason": f"{name} {ann_date} 审计意见异常：{result}（{agency}）。", + "source": "tushare.fina_audit", + }) + return alerts + + +def _classify_audit_opinion(result: str) -> tuple[str, bool]: + """Return (severity, reject) for non-standard audit opinions. + + Tushare often returns "标准无保留意见"; a plain substring check for + "保留意见" would incorrectly reject normal audit opinions. + """ + text = (result or "").strip() + if not text: + return "", False + normal_terms = ("标准无保留意见", "无保留意见") + warning_terms = ("带强调事项", "强调事项段", "非标准无保留") + reject_terms = ("无法表示", "否定意见", "保留意见", "非标") + if any(term in text for term in normal_terms) and not any(term in text for term in warning_terms): + return "", False + if any(term in text for term in ("无法表示", "否定意见", "非标")): + return "reject", True + if "保留意见" in text and not any(term in text for term in normal_terms): + return "reject", True + if any(term in text for term in warning_terms): + return "warning", False + return "", False + + +def _financial_statement_risks(ts_code: str, name: str, start: str, end: str) -> list[dict]: + df = tushare_client.get_balance_sheet(ts_code, start, end) + if df.empty: + return [] + row = df.iloc[-1] + total_assets = _float(row.get("total_assets")) + total_liab = _float(row.get("total_liab")) + goodwill = _float(row.get("goodwill")) + end_date = str(row.get("end_date") or "") + alerts = [] + if total_assets > 0: + goodwill_ratio = goodwill / total_assets * 100 + if goodwill_ratio >= settings.risk_goodwill_assets_warning_ratio: + reject = goodwill_ratio >= settings.risk_goodwill_assets_reject_ratio + alerts.append({ + "ts_code": ts_code, + "risk_type": "goodwill", + "severity": "reject" if reject else "warning", + "reject": reject, + "reason": f"{name} {end_date} 商誉/总资产约{goodwill_ratio:.1f}%，需警惕减值风险。", + "source": "tushare.balancesheet", + }) + debt_ratio = total_liab / total_assets * 100 + if debt_ratio >= settings.risk_debt_assets_warning_ratio: + reject = debt_ratio >= settings.risk_debt_assets_reject_ratio + alerts.append({ + "ts_code": ts_code, + "risk_type": "debt_pressure", + "severity": "reject" if reject else "warning", + "reject": reject, + "reason": f"{name} {end_date} 资产负债率约{debt_ratio:.1f}%，需关注偿债压力。", + "source": "tushare.balancesheet", + }) + return alerts + + +def _float(value: Any) -> float: + try: + if value in (None, ""): + return 0.0 + return float(value) + except Exception: + return 0.0 + + +def _dedupe_alerts(alerts: list[dict]) -> list[dict]: + seen = set() + result = [] + severity_rank = {"reject": 2, "warning": 1, "info": 0} + alerts = sorted(alerts, key=lambda item: (severity_rank.get(item.get("severity", ""), 0), bool(item.get("reject"))), reverse=True) + for item in alerts: + key = (item.get("ts_code"), item.get("risk_type"), item.get("reason")) + if key in seen: + continue + seen.add(key) + result.append(item) + return result diff --git a/backend/app/research/sector_agent.py b/backend/app/research/sector_agent.py new file mode 100644 index 00000000..4663057a --- /dev/null +++ b/backend/app/research/sector_agent.py @@ -0,0 +1,61 @@ +"""Sector and theme research agent.""" + +from __future__ import annotations + +from typing import Any + +from app.research.industry_chain_agent import ( + load_theme_chain_library, + map_sector_to_chain, + map_sector_to_chain_from_library, +) + + +def build_theme_views(sectors: list[Any]) -> list[dict]: + return _build_theme_views_with_mapper(sectors, map_sector_to_chain) + + +async def build_theme_views_async(sectors: list[Any]) -> list[dict]: + library = await load_theme_chain_library() + + def mapper(sector_name: str, leading_stocks: list[dict[str, Any]] | None = None) -> dict[str, Any]: + return map_sector_to_chain_from_library(sector_name, leading_stocks, library) + + return _build_theme_views_with_mapper(sectors, mapper) + + +def _build_theme_views_with_mapper(sectors: list[Any], mapper) -> list[dict]: + views: list[dict] = [] + for sector in sectors[:10]: + pct = getattr(sector, "realtime_pct_change", None) + if pct is None: + pct = getattr(sector, "pct_change", 0) or 0 + leading = getattr(sector, "leading_stocks_realtime", None) or getattr(sector, "leading_stocks", None) or [] + chain = mapper(getattr(sector, "sector_name", ""), leading) + stage = getattr(sector, "stage", "") or chain.get("stage") or "mid" + heat = float(getattr(sector, "heat_score", 0) or 0) + theme = chain["theme"] + logic = chain["logic"] + views.append({ + "theme": theme, + "raw_sector": getattr(sector, "sector_name", ""), + "stage": stage, + "heat_score": round(heat, 1), + "pct_change": round(float(pct or 0), 2), + "limit_up_count": int(getattr(sector, "limit_up_count", 0) or 0), + "logic": logic, + "chain_nodes": chain["chain_nodes"], + "chain_items": chain.get("chain_items", []), + "leader_stocks": leading[:5], + "lifecycle_status": chain.get("lifecycle_status") or _lifecycle_label(stage), + }) + return views + + +def _lifecycle_label(stage: str) -> str: + return { + "early": "启动期", + "mid": "扩散期", + "late": "后段", + "end": "退潮", + }.get(stage, "观察期") diff --git a/backend/app/research/stock_research_agent.py b/backend/app/research/stock_research_agent.py new file mode 100644 index 00000000..d899c448 --- /dev/null +++ b/backend/app/research/stock_research_agent.py @@ -0,0 +1,286 @@ +"""Stock research note generator. + +The stock research layer is explanatory. It enriches rule-selected candidates +with theme, catalyst and risk context, while deterministic notes remain the +fallback when the LLM or local catalyst data is unavailable. +""" + +from __future__ import annotations + +import json +import logging +import re +from typing import Any + +from sqlalchemy import text + +from app.config import settings +from app.db.database import get_db +from app.llm.client import chat_completion +from app.research.industry_chain_agent import infer_chain_node, infer_chain_position_from_theme_view + +logger = logging.getLogger(__name__) + + +def build_stock_research_notes_sync(recommendations: list[Any], theme_views: list[dict]) -> list[dict]: + """Build deterministic notes used by tests, fallback APIs and LLM failures.""" + theme_names = {item["theme"] for item in theme_views} + notes: list[dict] = [] + for rec in recommendations[:20]: + notes.append(_fallback_note(rec, theme_names, [], _match_theme_view(rec, theme_views))) + return notes + + +async def build_stock_research_notes( + recommendations: list[Any], + theme_views: list[dict], + risk_alerts: list[dict] | None = None, +) -> list[dict]: + """Build stock research notes, using LLM for top candidates when configured.""" + if not recommendations: + return [] + + theme_names = {item["theme"] for item in theme_views} + theme_map = {item["theme"]: item for item in theme_views} + risk_map = _group_risks(risk_alerts or []) + notes: list[dict] = [] + llm_limit = max(0, int(settings.research_stock_llm_limit or 0)) + + for index, rec in enumerate(recommendations[:20]): + theme = getattr(rec, "sector", "") or "未归类" + theme_view = _match_theme_view(rec, theme_views) + catalysts = await _load_local_catalyst_context(rec, theme) + fallback = _fallback_note(rec, theme_names, catalysts, theme_view) + if ( + not settings.research_stock_llm_enabled + or index >= llm_limit + or not settings.deepseek_api_key + ): + notes.append(fallback) + continue + + try: + llm_note = await _build_llm_note( + rec=rec, + fallback=fallback, + theme_view=theme_view or theme_map.get(theme, {}), + risks=risk_map.get(getattr(rec, "ts_code", ""), []), + catalysts=catalysts, + ) + notes.append(llm_note or fallback) + except Exception as exc: + logger.warning("股票研究笔记 LLM 生成失败 ts_code=%s error=%s", getattr(rec, "ts_code", ""), exc) + notes.append(fallback) + + return notes + + +def _fallback_note(rec: Any, theme_names: set[str], catalysts: list[dict], theme_view: dict | None = None) -> dict: + trace = getattr(rec, "decision_trace", {}) or {} + evidence = trace.get("evidence") or getattr(rec, "reasons", []) or [] + catalyst_titles = [item.get("title", "") for item in catalysts if item.get("title")] + theme = (theme_view or {}).get("theme") or getattr(rec, "sector", "") or "未归类" + position = ( + infer_chain_position_from_theme_view(theme_view, getattr(rec, "ts_code", ""), getattr(rec, "name", "")) + if theme_view + else {"chain_node": infer_chain_node(theme, getattr(rec, "name", ""), getattr(rec, "sector", "")), "stock_role": "待归类"} + ) + chain_node = position["chain_node"] + stock_role = position["stock_role"] + base = float(getattr(rec, "score", 0) or 0) + logic_score = min(100, base + (8 if theme in theme_names else 0) + min(len(catalyst_titles) * 1.5, 6)) + action = getattr(rec, "action_plan", "观察") or "观察" + invalid = getattr(rec, "invalidation_condition", "") or getattr(rec, "risk_note", "") or "板块热度回落、资金持续性不足或买点未触发。" + return { + "ts_code": getattr(rec, "ts_code", ""), + "name": getattr(rec, "name", ""), + "theme": theme, + "chain_node": chain_node, + "stock_role": stock_role, + "logic_score": round(logic_score, 1), + "logic_summary": f"{getattr(rec, 'name', '')} 属于 {theme} 方向，产业链位置为{chain_node}（{stock_role}），当前结论为{action}。", + "evidence": (evidence + catalyst_titles)[:5], + "uncertainty": getattr(rec, "risk_note", "") or "等待后续公告、资金持续性和板块生命周期验证。", + "disagreement": "若主线扩散失败、成交额无法维持或同板块核心股转弱，当前逻辑需要降级。", + "invalid_condition": invalid, + "generated_by": "rules", + } + + +async def _build_llm_note( + rec: Any, + fallback: dict, + theme_view: dict, + risks: list[dict], + catalysts: list[dict], +) -> dict | None: + payload = { + "stock": { + "ts_code": fallback["ts_code"], + "name": fallback["name"], + "theme": fallback["theme"], + "chain_node": fallback["chain_node"], + "stock_role": fallback.get("stock_role", "待归类"), + "score": getattr(rec, "score", 0), + "action_plan": getattr(rec, "action_plan", "观察"), + "trigger": getattr(rec, "trigger_condition", "") or getattr(rec, "entry_timing", ""), + "invalid_condition": getattr(rec, "invalidation_condition", "") or getattr(rec, "risk_note", ""), + "decision_trace": getattr(rec, "decision_trace", {}) or {}, + "reasons": getattr(rec, "reasons", []) or [], + }, + "theme_view": theme_view, + "recent_catalysts": catalysts, + "risk_alerts": risks, + } + messages = [ + { + "role": "system", + "content": ( + "你是A股研究员。你的任务是把系统筛出的候选标的整理成研究笔记，" + "用于解释机会逻辑、证据、分歧点和失效条件。不要给无条件买入建议，" + "不要编造未提供的数据。只输出合法JSON。" + ), + }, + { + "role": "user", + "content": ( + "请基于以下结构化输入生成研究笔记。输出JSON字段必须为：" + "logic_score(0-100数字), logic_summary(80字内), evidence(字符串数组,最多5条), " + "disagreement(80字内), uncertainty(80字内), invalid_condition(80字内)。\n\n" + f"{json.dumps(payload, ensure_ascii=False, default=str)}" + ), + }, + ] + message = await chat_completion(messages) + content = _message_content(message) + parsed = _extract_json_object(content) + if not parsed: + return None + + evidence = parsed.get("evidence") if isinstance(parsed.get("evidence"), list) else fallback["evidence"] + note = dict(fallback) + note.update({ + "logic_score": _clamp_float(parsed.get("logic_score"), fallback["logic_score"], 0, 100), + "logic_summary": _clean_text(parsed.get("logic_summary"), fallback["logic_summary"], 120), + "evidence": [_clean_text(item, "", 80) for item in evidence if _clean_text(item, "", 80)][:5], + "disagreement": _clean_text(parsed.get("disagreement"), fallback["disagreement"], 120), + "uncertainty": _clean_text(parsed.get("uncertainty"), fallback["uncertainty"], 120), + "invalid_condition": _clean_text(parsed.get("invalid_condition"), fallback["invalid_condition"], 120), + "generated_by": "llm", + }) + if not note["evidence"]: + note["evidence"] = fallback["evidence"] + return note + + +async def _load_local_catalyst_context(rec: Any, theme: str) -> list[dict]: + name = str(getattr(rec, "name", "") or "") + ts_code = str(getattr(rec, "ts_code", "") or "") + if not theme and not name and not ts_code: + return [] + + limit = max(1, int(settings.research_stock_news_limit or 6)) + params = { + "theme": theme, + "name": f"%{name}%" if name else "%", + "ts_code": f"%{ts_code}%" if ts_code else "%", + "limit": limit, + } + try: + async with get_db() as db: + result = await db.execute( + text( + "SELECT c.title, c.summary, c.source, c.published_at, c.catalyst_type, " + "c.strength, c.confidence, tc.theme_name, tc.reason " + "FROM catalysts c " + "LEFT JOIN theme_catalysts tc ON tc.catalyst_id = c.id " + "WHERE c.is_active = 1 AND (" + "tc.theme_name = :theme OR c.title LIKE :name OR c.summary LIKE :name OR c.raw_text LIKE :ts_code" + ") " + "ORDER BY COALESCE(c.published_at, c.created_at) DESC, c.id DESC " + "LIMIT :limit" + ), + params, + ) + rows = [] + for row in result.fetchall(): + item = dict(row._mapping) + rows.append({ + "title": item.get("title", ""), + "summary": item.get("summary", ""), + "source": item.get("source", ""), + "published_at": str(item.get("published_at") or ""), + "theme": item.get("theme_name", ""), + "reason": item.get("reason", ""), + "strength": item.get("strength", 0), + "confidence": item.get("confidence", 0), + }) + return rows + except Exception as exc: + logger.debug("读取本地催化上下文失败 ts_code=%s error=%s", ts_code, exc) + return [] + + +def _group_risks(risks: list[dict]) -> dict[str, list[dict]]: + grouped: dict[str, list[dict]] = {} + for risk in risks: + ts_code = str(risk.get("ts_code") or "") + if ts_code: + grouped.setdefault(ts_code, []).append(risk) + return grouped + + +def _match_theme_view(rec: Any, theme_views: list[dict]) -> dict: + sector = str(getattr(rec, "sector", "") or "") + for item in theme_views: + theme = str(item.get("theme") or "") + raw_sector = str(item.get("raw_sector") or "") + if sector and (sector == theme or sector == raw_sector or theme in sector or sector in raw_sector): + return item + return {} + + +def _message_content(message: Any) -> str: + if not message: + return "" + if isinstance(message, dict): + return str(message.get("content") or "") + return str(getattr(message, "content", "") or "") + + +def _extract_json_object(content: str) -> dict: + if not content: + return {} + cleaned = content.strip() + if cleaned.startswith("```"): + cleaned = re.sub(r"^```(?:json)?", "", cleaned, flags=re.IGNORECASE).strip() + cleaned = re.sub(r"```$", "", cleaned).strip() + try: + parsed = json.loads(cleaned) + return parsed if isinstance(parsed, dict) else {} + except Exception: + pass + + match = re.search(r"\{.*\}", cleaned, flags=re.DOTALL) + if not match: + return {} + try: + parsed = json.loads(match.group(0)) + return parsed if isinstance(parsed, dict) else {} + except Exception: + return {} + + +def _clean_text(value: Any, fallback: str, max_chars: int) -> str: + text_value = str(value or "").strip() + if not text_value: + text_value = fallback + return text_value[:max_chars] + + +def _clamp_float(value: Any, fallback: float, minimum: float, maximum: float) -> float: + try: + number = float(value) + except Exception: + number = float(fallback or 0) + return round(max(minimum, min(maximum, number)), 1) diff --git a/backend/astock.db b/backend/astock.db index 7ffd3270..558dadf6 100644 Binary files a/backend/astock.db and b/backend/astock.db differ diff --git a/frontend/src/app/(auth)/admin/page.tsx b/frontend/src/app/(auth)/admin/page.tsx index c0eac888..0c65e7da 100644 --- a/frontend/src/app/(auth)/admin/page.tsx +++ b/frontend/src/app/(auth)/admin/page.tsx @@ -85,6 +85,14 @@ export default function AdminPage() { secondary={`${stats.remainingInvites} 个剩余名额`} loading={loading} /> + ); diff --git a/frontend/src/app/(auth)/admin/themes/page.tsx b/frontend/src/app/(auth)/admin/themes/page.tsx new file mode 100644 index 00000000..f6b846b7 --- /dev/null +++ b/frontend/src/app/(auth)/admin/themes/page.tsx @@ -0,0 +1,316 @@ +"use client"; + +import Link from "next/link"; +import type { ReactNode } from "react"; +import { useCallback, useEffect, useMemo, useState } from "react"; +import { listThemeKnowledgeAPI, updateThemeKnowledgeAPI, type ThemeChainItem, type ThemeKnowledgeItem } from "@/lib/api"; +import { useAuth } from "@/hooks/use-auth"; + +const STAGES = [ + { value: "early", label: "启动期" }, + { value: "mid", label: "扩散期" }, + { value: "late", label: "后段" }, + { value: "end", label: "退潮" }, +]; + +export default function AdminThemesPage() { + const { user } = useAuth(); + const [items, setItems] = useState([]); + const [selected, setSelected] = useState(""); + const [draft, setDraft] = useState(null); + const [loading, setLoading] = useState(true); + const [message, setMessage] = useState(""); + + const loadData = useCallback(async () => { + if (user?.role !== "admin") return; + setLoading(true); + setMessage(""); + try { + const rows = await listThemeKnowledgeAPI(); + setItems(rows); + const first = rows[0]?.theme || ""; + setSelected(first); + setDraft(rows.find((item) => item.theme === first) ?? rows[0] ?? null); + } catch (error) { + setMessage(error instanceof Error ? error.message : "主题知识库加载失败"); + } finally { + setLoading(false); + } + }, [user?.role]); + + useEffect(() => { + loadData(); + }, [loadData]); + + const stats = useMemo(() => { + const nodes = items.reduce((sum, item) => sum + item.chain_nodes.length, 0); + const aliases = items.reduce((sum, item) => sum + item.aliases.length, 0); + return { themes: items.length, nodes, aliases }; + }, [items]); + + const selectTheme = (theme: string) => { + setSelected(theme); + setDraft(items.find((item) => item.theme === theme) ?? null); + setMessage(""); + }; + + const saveDraft = async () => { + if (!draft) return; + setMessage(""); + const saved = await updateThemeKnowledgeAPI(selected || draft.theme, draft); + setMessage(`${saved.theme} 已保存`); + await loadData(); + setSelected(saved.theme); + setDraft(saved); + }; + + if (user?.role !== "admin") { + return ; + } + + return ( +
+ + ← 返回系统管理 + + +
+
+
Theme Graph
+

主题知识库

+

维护主题、别名和产业链节点，作为研究报告和机会卡的基础语义层。

+
+ +
+ + {message ?
{message}
: null} + +
+ + + +
+ +
+
+
主题列表
+
+ {items.map((item) => ( + + ))} + {!items.length && !loading ?
暂无主题。
: null} +
+
+ + +
+
+ ); +} + +function ThemeEditor({ + draft, + setDraft, + onSave, +}: { + draft: ThemeKnowledgeItem | null; + setDraft: (item: ThemeKnowledgeItem) => void; + onSave: () => void; +}) { + if (!draft) { + return
选择一个主题后编辑。
; + } + + return ( +
+
+ + setDraft({ ...draft, theme: event.target.value })} + className="w-full rounded-xl border border-border-subtle bg-surface-2 px-3 py-2 text-sm text-text-primary outline-none focus:border-amber-500/40" + /> + + + + +
+ +
+ +