From f8020cab56dcedc07bd1a61a35fed647656e346b Mon Sep 17 00:00:00 2001 From: JiuContinent Date: Sun, 12 Jul 2026 22:05:23 +0800 Subject: [PATCH] =?UTF-8?q?```=20feat(market):=20=E6=B7=BB=E5=8A=A0?= =?UTF-8?q?=E5=B8=82=E5=9C=BA=E5=88=86=E6=9E=90=E6=95=B0=E6=8D=AE=E5=9F=BA?= =?UTF-8?q?=E7=A1=80=E6=9E=B6=E6=9E=84=E5=92=8C=E5=8A=9F=E8=83=BD=E6=A8=A1?= =?UTF-8?q?=E5=9D=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增市场分析相关数据库表结构，包括市场工具、每日报价、财务指标、 宏观指标和公告等数据模型 - 创建市场分析相关的Alembic迁移脚本，包含完整的up和downgrade逻辑 - 集成市场分析路由到主API路由器中 - 添加市场数据定时任务调度，支持盘前、收盘和周度市场分析报告 - 实现市场数据后台任务队列，包含报告生成和收盘分析功能 - 扩展系统配置设置，添加市场分析启用开关和相关参数配置 - 增加AI智能技能支持，包含市场概览分析和股票分析功能 - 添加审计日志记录，支持市场数据同步和自选股更新操作追踪 - 实现飞书命令集成，支持股票分析、自选股管理、公告查询等交互 - 提供市场数据服务层，包含行业分析、股票对比、市场概览等功能 ``` --- .../versions/202607120003_market_analysis.py | 122 ++ .../202607120004_market_announcements.py | 48 + app/api/router.py | 2 + app/core/background/scheduler.py | 42 + app/core/background/task_queue/__init__.py | 17 +- app/core/background/task_queue/constants.py | 2 + app/core/background/task_queue/legacy.py | 1 + app/core/background/task_queue/market.py | 26 + app/core/background/task_queue/reports.py | 1 + app/core/config/settings.py | 12 + app/modules/ai_agent/constants.py | 1 + app/modules/ai_agent/skills.py | 22 + app/modules/audit/constants.py | 5 + app/modules/business/models/__init__.py | 14 + app/modules/business/models/market.py | 87 ++ app/modules/feishu/commands.py | 262 ++++- app/modules/feishu/constants.py | 6 + app/modules/market/__init__.py | 3 + app/modules/market/chart.py | 62 + app/modules/market/pipeline.py | 263 +++++ app/modules/market/routes.py | 130 ++ app/modules/market/service.py | 1043 +++++++++++++++++ app/modules/reports/lifecycle_pipeline.py | 6 + app/modules/workflows/constants.py | 1 + app/tasks/__init__.py | 1 + app/tasks/market.py | 31 + app/tools/init_db.py | 12 + tests/test_smoke.py | 542 ++++++++- 28 files changed, 2704 insertions(+), 60 deletions(-) create mode 100644 alembic/versions/202607120003_market_analysis.py create mode 100644 alembic/versions/202607120004_market_announcements.py create mode 100644 app/core/background/task_queue/market.py create mode 100644 app/modules/business/models/market.py create mode 100644 app/modules/market/__init__.py create mode 100644 app/modules/market/chart.py create mode 100644 app/modules/market/pipeline.py create mode 100644 app/modules/market/routes.py create mode 100644 app/modules/market/service.py create mode 100644 app/tasks/market.py diff --git a/alembic/versions/202607120003_market_analysis.py b/alembic/versions/202607120003_market_analysis.py new file mode 100644 index 0000000..deae358 --- /dev/null +++ b/alembic/versions/202607120003_market_analysis.py @@ -0,0 +1,122 @@ +"""Add market analysis data foundation.""" + +from alembic import op +import sqlalchemy as sa + +revision = "202607120003" +down_revision = "202607120002" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "market_instruments", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("symbol", sa.String(32), nullable=False), + sa.Column("name", sa.String(128), nullable=False), + sa.Column("exchange", sa.String(16)), + sa.Column("instrument_type", sa.String(16), nullable=False), + sa.Column("industry", sa.String(128)), + sa.Column("list_date", sa.Date()), + sa.Column("is_active", sa.Boolean(), nullable=False), + sa.Column("source_updated_at", sa.DateTime()), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.Column("updated_at", sa.DateTime(), nullable=False), + ) + for column, unique in ( + ("symbol", True), + ("name", False), + ("exchange", False), + ("instrument_type", False), + ("industry", False), + ("is_active", False), + ): + op.create_index( + op.f(f"ix_market_instruments_{column}"), "market_instruments", [column], unique=unique + ) + op.create_table( + "market_daily_quotes", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("symbol", sa.String(32), nullable=False), + sa.Column("trade_date", sa.Date(), nullable=False), + sa.Column("open_price", sa.Numeric(18, 4)), + sa.Column("high_price", sa.Numeric(18, 4)), + sa.Column("low_price", sa.Numeric(18, 4)), + sa.Column("close_price", sa.Numeric(18, 4), nullable=False), + sa.Column("pre_close", sa.Numeric(18, 4)), + sa.Column("pct_change", sa.Numeric(12, 4)), + sa.Column("volume", sa.Numeric(20, 2)), + sa.Column("amount_cny", sa.Numeric(20, 2)), + sa.Column("pe", sa.Numeric(18, 4)), + sa.Column("pb", sa.Numeric(18, 4)), + sa.Column("total_market_value", sa.Numeric(20, 2)), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.Column("updated_at", sa.DateTime(), nullable=False), + sa.UniqueConstraint("symbol", "trade_date", name="uq_market_quote_day"), + ) + op.create_index(op.f("ix_market_daily_quotes_symbol"), "market_daily_quotes", ["symbol"]) + op.create_index( + op.f("ix_market_daily_quotes_trade_date"), "market_daily_quotes", ["trade_date"] + ) + op.create_table( + "market_financial_metrics", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("symbol", sa.String(32), nullable=False), + sa.Column("period_end", sa.Date(), nullable=False), + sa.Column("revenue_yoy", sa.Numeric(12, 4)), + sa.Column("net_profit_yoy", sa.Numeric(12, 4)), + sa.Column("roe", sa.Numeric(12, 4)), + sa.Column("debt_to_assets", sa.Numeric(12, 4)), + sa.Column("operating_cashflow_yoy", sa.Numeric(12, 4)), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.Column("updated_at", sa.DateTime(), nullable=False), + sa.UniqueConstraint("symbol", "period_end", name="uq_market_financial_period"), + ) + op.create_index( + op.f("ix_market_financial_metrics_symbol"), "market_financial_metrics", ["symbol"] + ) + op.create_index( + op.f("ix_market_financial_metrics_period_end"), "market_financial_metrics", ["period_end"] + ) + op.create_table( + "market_macro_indicators", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("code", sa.String(64), nullable=False), + sa.Column("name", sa.String(128), nullable=False), + sa.Column("period_date", sa.Date(), nullable=False), + sa.Column("value", sa.Numeric(20, 6), nullable=False), + sa.Column("unit", sa.String(32)), + sa.Column("source", sa.String(64), nullable=False), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.Column("updated_at", sa.DateTime(), nullable=False), + sa.UniqueConstraint("code", "period_date", name="uq_market_macro_period"), + ) + op.create_index(op.f("ix_market_macro_indicators_code"), "market_macro_indicators", ["code"]) + op.create_index( + op.f("ix_market_macro_indicators_period_date"), "market_macro_indicators", ["period_date"] + ) + op.create_table( + "market_watchlists", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("actor", sa.String(128), nullable=False), + sa.Column("symbol", sa.String(32), nullable=False), + sa.Column("enabled", sa.Boolean(), nullable=False), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.Column("updated_at", sa.DateTime(), nullable=False), + sa.UniqueConstraint("actor", "symbol", name="uq_market_watchlist_actor_symbol"), + ) + op.create_index(op.f("ix_market_watchlists_actor"), "market_watchlists", ["actor"]) + op.create_index(op.f("ix_market_watchlists_symbol"), "market_watchlists", ["symbol"]) + op.create_index(op.f("ix_market_watchlists_enabled"), "market_watchlists", ["enabled"]) + + +def downgrade() -> None: + for table in ( + "market_watchlists", + "market_macro_indicators", + "market_financial_metrics", + "market_daily_quotes", + "market_instruments", + ): + op.drop_table(table) diff --git a/alembic/versions/202607120004_market_announcements.py b/alembic/versions/202607120004_market_announcements.py new file mode 100644 index 0000000..4beeac0 --- /dev/null +++ b/alembic/versions/202607120004_market_announcements.py @@ -0,0 +1,48 @@ +"""Add market announcement metadata storage.""" + +from alembic import op +import sqlalchemy as sa + + +revision = "202607120004" +down_revision = "202607120003" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "market_announcements", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("source_key", sa.String(64), nullable=False), + sa.Column("symbol", sa.String(32)), + sa.Column("announcement_date", sa.Date(), nullable=False), + sa.Column("published_at", sa.DateTime()), + sa.Column("title", sa.String(512), nullable=False), + sa.Column("category", sa.String(128)), + sa.Column("source_url", sa.String(1024)), + sa.Column("source", sa.String(64), nullable=False), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.Column("updated_at", sa.DateTime(), nullable=False), + ) + op.create_index( + op.f("ix_market_announcements_source_key"), + "market_announcements", + ["source_key"], + unique=True, + ) + op.create_index( + op.f("ix_market_announcements_symbol"), "market_announcements", ["symbol"] + ) + op.create_index( + op.f("ix_market_announcements_announcement_date"), + "market_announcements", + ["announcement_date"], + ) + op.create_index( + op.f("ix_market_announcements_category"), "market_announcements", ["category"] + ) + + +def downgrade() -> None: + op.drop_table("market_announcements") diff --git a/app/api/router.py b/app/api/router.py index b1109e5..aee621f 100644 --- a/app/api/router.py +++ b/app/api/router.py @@ -9,6 +9,7 @@ from app.modules.dashboard.routes import router as dashboard_router from app.modules.events.routes import router as events_router from app.modules.feishu.routes import router as feishu_router from app.modules.legacy_mysql.routes import router as legacy_mysql_router +from app.modules.market.routes import router as market_router from app.modules.observability.routes import router as observability_router from app.modules.reports.routes import router as reports_router from app.modules.risk.routes import router as risk_router @@ -31,6 +32,7 @@ api_router.include_router(feishu_router, prefix="/integrations/feishu", tags=["f api_router.include_router(ai_router, prefix="/ai", tags=["ai"]) api_router.include_router(ai_memory_router, prefix="/ai", tags=["ai-memory"]) api_router.include_router(reports_router, prefix="/reports", tags=["reports"]) +api_router.include_router(market_router, prefix="/market", tags=["market"]) api_router.include_router(risk_router, prefix="/risks", tags=["risks"]) api_router.include_router(audit_router, prefix="/audit", tags=["audit"]) api_router.include_router(events_router, prefix="/events", tags=["events"]) diff --git a/app/core/background/scheduler.py b/app/core/background/scheduler.py index 9c4a016..0c7c766 100644 --- a/app/core/background/scheduler.py +++ b/app/core/background/scheduler.py @@ -1,3 +1,4 @@ +from datetime import date from socket import gethostname from typing import Any @@ -39,6 +40,7 @@ def create_scheduler(app: FastAPI | None = None) -> Any: enqueue_legacy_project_sync, enqueue_legacy_task_sync, enqueue_lifecycle_report, + enqueue_market_report, enqueue_project_weekly_push, ) from app.modules.observability.service import ObservabilityService @@ -127,6 +129,18 @@ def create_scheduler(app: FastAPI | None = None) -> Any: ) _set_state(app, "last_weekly_lifecycle_dispatch", dispatch) + def run_market_premarket() -> None: + dispatch = enqueue_market_report("premarket") + _set_state(app, "last_market_premarket_dispatch", dispatch) + + def run_market_close() -> None: + dispatch = enqueue_market_report("close") + _set_state(app, "last_market_close_dispatch", dispatch) + + def run_market_weekly() -> None: + dispatch = enqueue_market_report("weekly", date.today()) + _set_state(app, "last_market_weekly_dispatch", dispatch) + def run_event_dispatch() -> None: dispatch = enqueue_event_dispatch( limit=settings.event_dispatch_batch_size, @@ -190,6 +204,34 @@ def create_scheduler(app: FastAPI | None = None) -> Any: id="event_dispatch", replace_existing=True, ) + if settings.market_analysis_enabled: + scheduler.add_job( + run_market_premarket, + trigger="cron", + day_of_week="mon-fri", + hour=settings.market_premarket_cron_hour, + minute=settings.market_premarket_cron_minute, + id="market_premarket_analysis", + replace_existing=True, + ) + scheduler.add_job( + run_market_close, + trigger="cron", + day_of_week="mon-fri", + hour=settings.market_close_cron_hour, + minute=settings.market_close_cron_minute, + id="market_close_analysis", + replace_existing=True, + ) + scheduler.add_job( + run_market_weekly, + trigger="cron", + day_of_week=settings.market_weekly_day_of_week, + hour=settings.market_weekly_cron_hour, + minute=settings.market_weekly_cron_minute, + id="market_weekly_analysis", + replace_existing=True, + ) scheduler.add_job( record_scheduler_heartbeat, trigger="interval", diff --git a/app/core/background/task_queue/__init__.py b/app/core/background/task_queue/__init__.py index d3d4470..f2f39d9 100644 --- a/app/core/background/task_queue/__init__.py +++ b/app/core/background/task_queue/__init__.py @@ -6,12 +6,21 @@ from app.core.background.task_queue.constants import ( TASK_SYNC_LEGACY_PROJECTS, TASK_SYNC_LEGACY_TASKS, TASK_RUN_LIFECYCLE, + TASK_RUN_MARKET_CLOSE, + TASK_RUN_MARKET_REPORT, ) from app.core.background.task_queue.dispatcher import dispatch_task from app.core.background.task_queue.events import enqueue_event_dispatch -from app.core.background.task_queue.legacy import enqueue_legacy_project_sync, enqueue_legacy_task_sync +from app.core.background.task_queue.legacy import ( + enqueue_legacy_project_sync, + enqueue_legacy_task_sync, +) from app.core.background.task_queue.lifecycle import enqueue_lifecycle_report -from app.core.background.task_queue.reports import enqueue_daily_brief_push, enqueue_project_weekly_push +from app.core.background.task_queue.market import enqueue_market_close, enqueue_market_report +from app.core.background.task_queue.reports import ( + enqueue_daily_brief_push, + enqueue_project_weekly_push, +) from app.core.background.task_queue.risk import enqueue_risk_event_generation @@ -23,12 +32,16 @@ __all__ = [ "TASK_SYNC_LEGACY_PROJECTS", "TASK_SYNC_LEGACY_TASKS", "TASK_RUN_LIFECYCLE", + "TASK_RUN_MARKET_CLOSE", + "TASK_RUN_MARKET_REPORT", "dispatch_task", "enqueue_daily_brief_push", "enqueue_event_dispatch", "enqueue_legacy_project_sync", "enqueue_legacy_task_sync", "enqueue_lifecycle_report", + "enqueue_market_close", + "enqueue_market_report", "enqueue_project_weekly_push", "enqueue_risk_event_generation", ] diff --git a/app/core/background/task_queue/constants.py b/app/core/background/task_queue/constants.py index 5952789..67c5bee 100644 --- a/app/core/background/task_queue/constants.py +++ b/app/core/background/task_queue/constants.py @@ -5,3 +5,5 @@ TASK_SYNC_LEGACY_PROJECTS = "legacy.sync_projects" TASK_SYNC_LEGACY_TASKS = "legacy.sync_tasks" TASK_DISPATCH_PENDING_EVENTS = "events.dispatch_pending" TASK_RUN_LIFECYCLE = "reports.run_lifecycle" +TASK_RUN_MARKET_REPORT = "market.report.run" +TASK_RUN_MARKET_CLOSE = TASK_RUN_MARKET_REPORT diff --git a/app/core/background/task_queue/legacy.py b/app/core/background/task_queue/legacy.py index 80578a3..0173ecd 100644 --- a/app/core/background/task_queue/legacy.py +++ b/app/core/background/task_queue/legacy.py @@ -45,6 +45,7 @@ def enqueue_legacy_project_sync( inline, ) + def enqueue_legacy_task_sync( source_query: str | None = None, source_query_name: str | None = None, diff --git a/app/core/background/task_queue/market.py b/app/core/background/task_queue/market.py new file mode 100644 index 0000000..6d0d062 --- /dev/null +++ b/app/core/background/task_queue/market.py @@ -0,0 +1,26 @@ +from datetime import date +from typing import Any + +from app.core.background.task_queue.constants import TASK_RUN_MARKET_REPORT +from app.core.config import get_settings + + +def enqueue_market_report( + report_type: str, trade_date: date | None = None, force: bool = False +) -> dict[str, Any]: + value = trade_date.isoformat() if trade_date else None + if not get_settings().task_queue_enabled: + from app.tasks.market import run_market_report + + return run_market_report.run(report_type, value, force) + from app.tasks import celery_app + + result = celery_app.signature( + TASK_RUN_MARKET_REPORT, + kwargs={"report_type": report_type, "trade_date": value, "force": force}, + ).apply_async() + return {"task_id": result.id, "status": "queued"} + + +def enqueue_market_close(trade_date: date | None = None) -> dict[str, Any]: + return enqueue_market_report("close", trade_date) diff --git a/app/core/background/task_queue/reports.py b/app/core/background/task_queue/reports.py index ef8424b..765952a 100644 --- a/app/core/background/task_queue/reports.py +++ b/app/core/background/task_queue/reports.py @@ -76,6 +76,7 @@ def enqueue_daily_brief_push( inline, ) + def enqueue_project_weekly_push( receive_id: str | None = None, receive_id_type: str = FeishuReceiveIdType.CHAT_ID, diff --git a/app/core/config/settings.py b/app/core/config/settings.py index 3cf23c2..751e7e8 100644 --- a/app/core/config/settings.py +++ b/app/core/config/settings.py @@ -71,6 +71,17 @@ class Settings(BaseSettings): task_queue_always_eager: bool = False lifecycle_pipeline_enabled: bool = False finance_needs_enabled: bool = False + market_analysis_enabled: bool = False + market_data_provider: str = "tushare" + market_data_base_url: str = "https://api.tushare.pro" + market_data_token: str | None = None + market_premarket_cron_hour: int = 8 + market_premarket_cron_minute: int = 30 + market_close_cron_hour: int = 15 + market_close_cron_minute: int = 30 + market_weekly_day_of_week: str = "sun" + market_weekly_cron_hour: int = 20 + market_weekly_cron_minute: int = 0 legacy_sync_enabled: bool = False celery_result_backend_url: str | None = None daily_brief_cron_hour: int = 9 @@ -106,6 +117,7 @@ class Settings(BaseSettings): "openclaw_gateway_token", "hermes_api_key", "direct_llm_api_key", + "market_data_token", "feishu_app_secret", "feishu_verification_token", ] diff --git a/app/modules/ai_agent/constants.py b/app/modules/ai_agent/constants.py index d0e6e87..1f8fce9 100644 --- a/app/modules/ai_agent/constants.py +++ b/app/modules/ai_agent/constants.py @@ -166,5 +166,6 @@ AI_AUDIT_SENSITIVE_KEYS = frozenset( "openclaw_gateway_token", "hermes_api_key", "direct_llm_api_key", + "market_data_token", } ) diff --git a/app/modules/ai_agent/skills.py b/app/modules/ai_agent/skills.py index c571df3..1e6055d 100644 --- a/app/modules/ai_agent/skills.py +++ b/app/modules/ai_agent/skills.py @@ -14,6 +14,8 @@ class AISkillId(StrEnum): INVESTMENT_RESEARCH = "investment_research" PROJECT_LIFECYCLE_ANALYSIS = "project_lifecycle_analysis" PROJECT_FINANCE_NEEDS_ANALYSIS = "project_finance_needs_analysis" + MARKET_OVERVIEW_ANALYSIS = "market_overview_analysis" + STOCK_ANALYSIS = "stock_analysis" HERMES_MEMORY_RECALL = "hermes_memory_recall" HERMES_MEMORY_WRITE = "hermes_memory_write" @@ -68,6 +70,16 @@ PROJECT_FINANCE_NEEDS_ANALYSIS_INSTRUCTIONS = ( "完成时限和验证指标；五、数据限制。所有安排必须标注需要人工确认。" "不得把项目资金安排需求称为公司融资缺口，不得审批付款、融资或投资交易。" ) +MARKET_OVERVIEW_ANALYSIS_INSTRUCTIONS = ( + "基于上下文中的A股指数、涨跌家数、成交额和行业数据生成中文市场研究。" + "输出总体判断、证据、前三项风险、关注行业、下一交易日观察指标和数据限制。" + "不得编造行情、保证收益或下交易指令。" +) +STOCK_ANALYSIS_INSTRUCTIONS = ( + "基于上下文中的价格、收益、均线、波动率、回撤、估值和财务指标生成中文股票研究。" + "输出基本面、技术面、估值、三种情景、风险及后续跟踪指标。" + "不得给出保证性结论或自动交易指令，所有决策必须人工确认。" +) HERMES_MEMORY_RECALL_INSTRUCTIONS = ( "Retrieve concise long-term memory, preferences, prior decisions, and relevant " "business context for this request. Return only information useful to answer it." @@ -98,6 +110,16 @@ AI_SKILLS: dict[AISkillId, AISkill] = { source=AISkillSource.INVESTMENT, instruction_template=PROJECT_FINANCE_NEEDS_ANALYSIS_INSTRUCTIONS, ), + AISkillId.MARKET_OVERVIEW_ANALYSIS: AISkill( + skill_id=AISkillId.MARKET_OVERVIEW_ANALYSIS, + source=AISkillSource.INVESTMENT, + instruction_template=MARKET_OVERVIEW_ANALYSIS_INSTRUCTIONS, + ), + AISkillId.STOCK_ANALYSIS: AISkill( + skill_id=AISkillId.STOCK_ANALYSIS, + source=AISkillSource.INVESTMENT, + instruction_template=STOCK_ANALYSIS_INSTRUCTIONS, + ), AISkillId.HERMES_MEMORY_RECALL: AISkill( skill_id=AISkillId.HERMES_MEMORY_RECALL, source=AISkillSource.AI_MEMORY, diff --git a/app/modules/audit/constants.py b/app/modules/audit/constants.py index b92a0f6..9bf845f 100644 --- a/app/modules/audit/constants.py +++ b/app/modules/audit/constants.py @@ -22,6 +22,8 @@ class AuditAction(StrEnum): ENTERPRISE_ANALYTICS = "enterprise_analytics" EVENT_DISPATCH = "event.dispatch" HEARTBEAT = "heartbeat" + MARKET_SYNC = "market.sync" + MARKET_WATCHLIST_UPDATE = "market.watchlist.update" class AuditRiskLevel(StrEnum): @@ -40,6 +42,7 @@ class AuditSource(StrEnum): EVENTS = "events" AI_MEMORY = "ai_memory" OBSERVABILITY = "observability" + MARKET = "market" class AuditTargetType(StrEnum): @@ -51,6 +54,7 @@ class AuditTargetType(StrEnum): AI_MEMORY = "ai-memory" DOMAIN_EVENT = "domain-event" HEARTBEAT = "heartbeat" + MARKET = "market" class AuditStatus(StrEnum): @@ -71,6 +75,7 @@ AUDIT_SENSITIVE_KEYS = frozenset( "openclaw_gateway_token", "hermes_api_key", "direct_llm_api_key", + "market_data_token", "feishu_app_secret", "feishu_verification_token", } diff --git a/app/modules/business/models/__init__.py b/app/modules/business/models/__init__.py index 9478a1e..535c63f 100644 --- a/app/modules/business/models/__init__.py +++ b/app/modules/business/models/__init__.py @@ -2,6 +2,14 @@ from app.modules.business.models.attendance import AttendanceRecord from app.modules.business.models.finance import Expense, FundAccount, Procurement from app.modules.business.models.governance import PerformanceMetric, Policy, Standard from app.modules.business.models.legacy import LegacySyncRun +from app.modules.business.models.market import ( + MarketAnnouncement, + MarketDailyQuote, + MarketFinancialMetric, + MarketInstrument, + MarketMacroIndicator, + MarketWatchlist, +) from app.modules.business.models.lifecycle import ( Employee, ProjectCashFlow, @@ -22,6 +30,12 @@ __all__ = [ "Employee", "FundAccount", "LegacySyncRun", + "MarketAnnouncement", + "MarketDailyQuote", + "MarketFinancialMetric", + "MarketInstrument", + "MarketMacroIndicator", + "MarketWatchlist", "PerformanceMetric", "Policy", "Procurement", diff --git a/app/modules/business/models/market.py b/app/modules/business/models/market.py new file mode 100644 index 0000000..5346a62 --- /dev/null +++ b/app/modules/business/models/market.py @@ -0,0 +1,87 @@ +from datetime import date, datetime +from decimal import Decimal + +from sqlalchemy import Boolean, Date, DateTime, Integer, Numeric, String, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column + +from app.core.database import Base +from app.modules.business.models.common import TimestampMixin + + +class MarketInstrument(Base, TimestampMixin): + __tablename__ = "market_instruments" + id: Mapped[int] = mapped_column(Integer, primary_key=True) + symbol: Mapped[str] = mapped_column(String(32), unique=True, index=True) + name: Mapped[str] = mapped_column(String(128), index=True) + exchange: Mapped[str | None] = mapped_column(String(16), nullable=True, index=True) + instrument_type: Mapped[str] = mapped_column(String(16), default="stock", index=True) + industry: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True) + list_date: Mapped[date | None] = mapped_column(Date, nullable=True) + is_active: Mapped[bool] = mapped_column(Boolean, default=True, index=True) + source_updated_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + + +class MarketDailyQuote(Base, TimestampMixin): + __tablename__ = "market_daily_quotes" + __table_args__ = (UniqueConstraint("symbol", "trade_date", name="uq_market_quote_day"),) + id: Mapped[int] = mapped_column(Integer, primary_key=True) + symbol: Mapped[str] = mapped_column(String(32), index=True) + trade_date: Mapped[date] = mapped_column(Date, index=True) + open_price: Mapped[Decimal | None] = mapped_column(Numeric(18, 4), nullable=True) + high_price: Mapped[Decimal | None] = mapped_column(Numeric(18, 4), nullable=True) + low_price: Mapped[Decimal | None] = mapped_column(Numeric(18, 4), nullable=True) + close_price: Mapped[Decimal] = mapped_column(Numeric(18, 4)) + pre_close: Mapped[Decimal | None] = mapped_column(Numeric(18, 4), nullable=True) + pct_change: Mapped[Decimal | None] = mapped_column(Numeric(12, 4), nullable=True) + volume: Mapped[Decimal | None] = mapped_column(Numeric(20, 2), nullable=True) + amount_cny: Mapped[Decimal | None] = mapped_column(Numeric(20, 2), nullable=True) + pe: Mapped[Decimal | None] = mapped_column(Numeric(18, 4), nullable=True) + pb: Mapped[Decimal | None] = mapped_column(Numeric(18, 4), nullable=True) + total_market_value: Mapped[Decimal | None] = mapped_column(Numeric(20, 2), nullable=True) + + +class MarketFinancialMetric(Base, TimestampMixin): + __tablename__ = "market_financial_metrics" + __table_args__ = (UniqueConstraint("symbol", "period_end", name="uq_market_financial_period"),) + id: Mapped[int] = mapped_column(Integer, primary_key=True) + symbol: Mapped[str] = mapped_column(String(32), index=True) + period_end: Mapped[date] = mapped_column(Date, index=True) + revenue_yoy: Mapped[Decimal | None] = mapped_column(Numeric(12, 4), nullable=True) + net_profit_yoy: Mapped[Decimal | None] = mapped_column(Numeric(12, 4), nullable=True) + roe: Mapped[Decimal | None] = mapped_column(Numeric(12, 4), nullable=True) + debt_to_assets: Mapped[Decimal | None] = mapped_column(Numeric(12, 4), nullable=True) + operating_cashflow_yoy: Mapped[Decimal | None] = mapped_column(Numeric(12, 4), nullable=True) + + +class MarketMacroIndicator(Base, TimestampMixin): + __tablename__ = "market_macro_indicators" + __table_args__ = (UniqueConstraint("code", "period_date", name="uq_market_macro_period"),) + id: Mapped[int] = mapped_column(Integer, primary_key=True) + code: Mapped[str] = mapped_column(String(64), index=True) + name: Mapped[str] = mapped_column(String(128)) + period_date: Mapped[date] = mapped_column(Date, index=True) + value: Mapped[Decimal] = mapped_column(Numeric(20, 6)) + unit: Mapped[str | None] = mapped_column(String(32), nullable=True) + source: Mapped[str] = mapped_column(String(64), default="tushare") + + +class MarketAnnouncement(Base, TimestampMixin): + __tablename__ = "market_announcements" + id: Mapped[int] = mapped_column(Integer, primary_key=True) + source_key: Mapped[str] = mapped_column(String(64), unique=True, index=True) + symbol: Mapped[str | None] = mapped_column(String(32), nullable=True, index=True) + announcement_date: Mapped[date] = mapped_column(Date, index=True) + published_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + title: Mapped[str] = mapped_column(String(512)) + category: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True) + source_url: Mapped[str | None] = mapped_column(String(1024), nullable=True) + source: Mapped[str] = mapped_column(String(64), default="tushare") + + +class MarketWatchlist(Base, TimestampMixin): + __tablename__ = "market_watchlists" + __table_args__ = (UniqueConstraint("actor", "symbol", name="uq_market_watchlist_actor_symbol"),) + id: Mapped[int] = mapped_column(Integer, primary_key=True) + actor: Mapped[str] = mapped_column(String(128), index=True) + symbol: Mapped[str] = mapped_column(String(32), index=True) + enabled: Mapped[bool] = mapped_column(Boolean, default=True, index=True) diff --git a/app/modules/feishu/commands.py b/app/modules/feishu/commands.py index 0ae696e..6b9f25f 100644 --- a/app/modules/feishu/commands.py +++ b/app/modules/feishu/commands.py @@ -26,6 +26,8 @@ from app.modules.reports.constants import ReportResponseKey from app.modules.reports.chart import lifecycle_chart_alt, render_lifecycle_chart from app.modules.feishu.service import FeishuService from app.modules.reports.services import ReportService +from app.modules.market.chart import render_market_chart +from app.modules.market.service import MarketService from app.modules.risk.constants import RiskSummaryKey from app.modules.risk.services import RiskService @@ -38,20 +40,48 @@ RISK_TITLE = "风险预警" DEFAULT_AI_PROMPT = "请说明你能做什么。" RULE_TITLE = "AI 学习规则" RULE_CREATE_PATTERN = re.compile(r"^学习规则(?:\s+(\d{1,3}))?\s*[：:]\s*(.*)$") +MARKET_RULE_CREATE_PATTERN = re.compile( + r"^学习市场规则(?:\s+(\d{1,3}))?\s*[：:]\s*(.*)$" +) RULE_DISABLE_PATTERN = re.compile(r"^停用规则\s+(MEM-[A-Za-z0-9-]+)$", re.IGNORECASE) RULE_ENABLE_PATTERN = re.compile(r"^启用规则\s+(MEM-[A-Za-z0-9-]+)$", re.IGNORECASE) -RULE_LIST_COMMANDS = {"查看规则", "规则列表"} -RULE_COMMAND_PREFIXES = ("学习规则", "查看规则", "规则列表", "停用规则", "启用规则") +RULE_LIST_COMMANDS = {"查看规则", "规则列表", "查看市场规则"} +RULE_COMMAND_PREFIXES = ( + "学习市场规则", + "学习规则", + "查看市场规则", + "查看规则", + "规则列表", + "停用规则", + "启用规则", +) RULE_COMMAND_HELP = ( "规则指令格式：\n" "学习规则：<规则内容>\n" "学习规则 80：<规则内容>\n" + "学习市场规则 80：<仅用于市场分析的规则内容>\n" "查看规则\n" "停用规则 <规则编号>\n" "启用规则 <规则编号>" ) PROJECT_FINANCE_PATTERN = re.compile(r"^项目资金\s+(.+)$") FINANCE_COMMANDS = {"资金需求", "未来30天资金需求"} +STOCK_ANALYSIS_PATTERN = re.compile( + r"^(?:股票分析|估值分析|财报分析)\s+([0-9]{6}(?:\.(?:SH|SZ|BJ))?)$", re.IGNORECASE +) +WATCHLIST_ADD_PATTERN = re.compile(r"^加入自选\s+([0-9]{6}(?:\.(?:SH|SZ|BJ))?)$", re.IGNORECASE) +MARKET_COMMANDS = { + "市场分析", + "今日收盘分析", + "本周市场分析", + "宏观金融分析", + "最新公告", +} +INDUSTRY_ANALYSIS_PATTERN = re.compile(r"^行业分析\s+(.+)$") +STOCK_COMPARE_PATTERN = re.compile( + r"^股票对比\s+([0-9]{6}(?:\.(?:SH|SZ|BJ))?)\s+" r"([0-9]{6}(?:\.(?:SH|SZ|BJ))?)$", + re.IGNORECASE, +) def _parse_content_text(content: Any) -> str: @@ -59,9 +89,7 @@ def _parse_content_text(content: Any) -> str: if isinstance(content, dict): return str( - content.get(FeishuPayloadKey.TEXT) - or content.get(FeishuPayloadKey.CONTENT) - or "" + content.get(FeishuPayloadKey.TEXT) or content.get(FeishuPayloadKey.CONTENT) or "" ) if not isinstance(content, str): return "" @@ -70,11 +98,7 @@ def _parse_content_text(content: Any) -> str: except json.JSONDecodeError: return content if isinstance(data, dict): - return str( - data.get(FeishuPayloadKey.TEXT) - or data.get(FeishuPayloadKey.CONTENT) - or "" - ) + return str(data.get(FeishuPayloadKey.TEXT) or data.get(FeishuPayloadKey.CONTENT) or "") return content @@ -163,6 +187,10 @@ class FeishuCommandService: if finance_result is not None: return finance_result + market_result = self._handle_market_command(command_text, chat_id, actor, auto_reply) + if market_result is not None: + return market_result + if any(keyword in command_text for keyword in DAILY_REPORT_KEYWORDS): report = ReportService(self.db).daily_brief() if auto_reply: @@ -284,9 +312,7 @@ class FeishuCommandService: if command_text not in FINANCE_COMMANDS and project_match is None: return None command = ( - FeishuCommandName.PROJECT_FINANCE - if project_match - else FeishuCommandName.FINANCE_NEEDS + FeishuCommandName.PROJECT_FINANCE if project_match else FeishuCommandName.FINANCE_NEEDS ) if not get_settings().finance_needs_enabled: content = "项目资金需求分析尚未启用，请先配置并启用财务只读同步。" @@ -362,6 +388,184 @@ class FeishuCommandService: report[ReportResponseKey.LINES], ) + def _handle_market_command( + self, text: str, chat_id: str | None, actor: str, auto_reply: bool + ) -> dict[str, Any] | None: + stock = STOCK_ANALYSIS_PATTERN.fullmatch(text) + add = WATCHLIST_ADD_PATTERN.fullmatch(text) + industry = INDUSTRY_ANALYSIS_PATTERN.fullmatch(text) + comparison = STOCK_COMPARE_PATTERN.fullmatch(text) + if ( + text not in MARKET_COMMANDS + and text != "查看自选" + and not stock + and not add + and not industry + and not comparison + ): + return None + if not get_settings().market_analysis_enabled: + content = "市场分析尚未启用，请配置市场数据源后启用。" + response = ( + self._send_text_if_configured(chat_id, content, actor) if auto_reply else None + ) + return _command_result( + FeishuCommandName.MARKET_OVERVIEW, + FeishuReplyType.TEXT, + "市场分析", + content, + response, + ) + service = MarketService(self.db) + if add: + if get_settings().read_only_mode: + content = "当前为只读模式，不能修改自选股。请由管理员启用操作后重试。" + response = ( + self._send_text_if_configured(chat_id, content, actor) + if auto_reply + else None + ) + return _command_result( + FeishuCommandName.WATCHLIST_ADD, + FeishuReplyType.TEXT, + "自选股", + content, + response, + ) + item = service.add_watchlist(actor, add.group(1)) + content = f"已加入自选：{item['symbol']}" + response = ( + self._send_text_if_configured(chat_id, content, actor) if auto_reply else None + ) + return _command_result( + FeishuCommandName.WATCHLIST_ADD, FeishuReplyType.TEXT, "自选股", content, response + ) + if text == "查看自选": + items = service.watchlist(actor) + content = "自选股：" + ("、".join(item["symbol"] for item in items) or "暂无") + response = ( + self._send_text_if_configured(chat_id, content, actor) if auto_reply else None + ) + return _command_result( + FeishuCommandName.WATCHLIST_LIST, FeishuReplyType.TEXT, "自选股", content, response + ) + if text == "最新公告": + items = service.announcements(limit=10)["items"] + content = ( + "最新公告：\n" + + "\n".join( + f"- {item['announcement_date']} {item['symbol'] or '市场'}：{item['title']}" + for item in items + ) + if items + else "公告元数据尚未接入。" + ) + response = ( + self._send_text_if_configured(chat_id, content, actor) if auto_reply else None + ) + return _command_result( + FeishuCommandName.MARKET_ANNOUNCEMENTS, + FeishuReplyType.TEXT, + "最新公告", + content, + response, + ) + if industry: + try: + data = service.industry_analysis(industry.group(1).strip()) + content = ( + f"{data['industry']} 平均涨跌 {data['average_pct_change']}%\n" + + "\n".join( + f"- {item['name']}（{item['symbol']}）：{item['pct_change']}%" + for item in data["items"][:10] + ) + ) + except HTTPException: + content = "未找到该行业的最新市场数据。" + response = ( + self._send_text_if_configured(chat_id, content, actor) if auto_reply else None + ) + return _command_result( + FeishuCommandName.MARKET_OVERVIEW, + FeishuReplyType.TEXT, + "行业分析", + content, + response, + ) + if comparison: + try: + content = service.compare_stocks([comparison.group(1), comparison.group(2)])[ + "content" + ] + except HTTPException: + content = "至少一只股票缺少可用行情，暂时无法比较。" + response = ( + self._send_text_if_configured(chat_id, content, actor) if auto_reply else None + ) + return _command_result( + FeishuCommandName.STOCK_ANALYSIS, + FeishuReplyType.TEXT, + "股票对比", + content, + response, + ) + command = FeishuCommandName.STOCK_ANALYSIS if stock else FeishuCommandName.MARKET_OVERVIEW + if text == "宏观金融分析": + command = FeishuCommandName.MARKET_MACRO + try: + if stock: + report = service.stock_analysis(stock.group(1), True, actor) + elif text == "本周市场分析": + report = service.weekly_overview(include_ai=True, actor=actor) + elif text == "宏观金融分析": + report = service.macro_analysis(include_ai=True, actor=actor) + else: + report = service.market_overview(include_ai=True, actor=actor) + except HTTPException: + content = "未找到该股票的可用行情，请确认代码或先执行行情同步。" + response = ( + self._send_text_if_configured(chat_id, content, actor) if auto_reply else None + ) + return _command_result(command, FeishuReplyType.TEXT, "股票分析", content, response) + ai = report.get("ai_analysis") or {} + if not report.get("data_available"): + content = "市场数据未接入，暂不生成分析报告。" + elif not ai.get("ok"): + content = "AI 当前不可用，本次市场分析报告未发送。" + else: + content = report["content"] + response = None + if auto_reply: + if ai.get("ok"): + if text == "宏观金融分析": + response = self._send_text_if_configured(chat_id, content, actor) + else: + image = self.feishu.upload_image(render_market_chart(report), actor) + image_key = (image.get("data") or {}).get("image_key") + if not image_key: + raise ValueError("Feishu image upload did not return image_key") + card = FeishuService.build_basic_card( + report["title"], + report["lines"], + image_key=image_key, + image_alt=report["title"], + ) + response = self.feishu.send_card(card, receive_id=chat_id, actor=actor) + else: + response = self._send_text_if_configured(chat_id, content, actor) + return _command_result( + command, + ( + FeishuReplyType.CARD + if ai.get("ok") and text != "宏观金融分析" + else FeishuReplyType.TEXT + ), + report["title"], + content, + response, + report["lines"] if ai.get("ok") else None, + ) + def _send_finance_card_if_configured( self, chat_id: str | None, @@ -401,13 +605,30 @@ class FeishuCommandService: command = FeishuCommandName.RULE_DISABLE elif command_text.startswith("启用规则"): command = FeishuCommandName.RULE_ENABLE - elif command_text.startswith(("查看规则", "规则列表")): + elif command_text.startswith(("查看市场规则", "查看规则", "规则列表")): command = FeishuCommandName.RULE_LIST else: command = FeishuCommandName.RULE_CREATE + if command in { + FeishuCommandName.RULE_CREATE, + FeishuCommandName.RULE_DISABLE, + FeishuCommandName.RULE_ENABLE, + } and get_settings().read_only_mode: + content = "当前为只读模式，不能新增或修改学习规则。请由管理员启用操作后重试。" + provider_response = ( + self._send_text_if_configured(chat_id, content, actor) if auto_reply else None + ) + return _command_result( + command, + FeishuReplyType.TEXT, + RULE_TITLE, + content, + provider_response, + ) content = RULE_COMMAND_HELP try: - create_match = RULE_CREATE_PATTERN.fullmatch(command_text) + market_create_match = MARKET_RULE_CREATE_PATTERN.fullmatch(command_text) + create_match = market_create_match or RULE_CREATE_PATTERN.fullmatch(command_text) disable_match = RULE_DISABLE_PATTERN.fullmatch(command_text) enable_match = RULE_ENABLE_PATTERN.fullmatch(command_text) memory = AIMemoryService(self.db) @@ -422,10 +643,10 @@ class FeishuCommandService: else: rule = memory.create_rule( content=rule_content, - scope="global", - subject="company", + scope="market" if market_create_match else "global", + subject="market" if market_create_match else "company", priority=priority, - tags=["feishu"], + tags=["feishu", *(["market"] if market_create_match else [])], actor=actor, ) content = ( @@ -438,6 +659,7 @@ class FeishuCommandService: elif command_text in RULE_LIST_COMMANDS: command = FeishuCommandName.RULE_LIST rules = memory.list_rules( + scope="market" if command_text == "查看市场规则" else None, status_filter=AIMemoryStatus.ACTIVE, limit=20, ) @@ -457,9 +679,7 @@ class FeishuCommandService: elif disable_match or enable_match: enabled = enable_match is not None command = ( - FeishuCommandName.RULE_ENABLE - if enabled - else FeishuCommandName.RULE_DISABLE + FeishuCommandName.RULE_ENABLE if enabled else FeishuCommandName.RULE_DISABLE ) match = enable_match or disable_match rule = memory.update_rule( diff --git a/app/modules/feishu/constants.py b/app/modules/feishu/constants.py index 55a1172..ed158b4 100644 --- a/app/modules/feishu/constants.py +++ b/app/modules/feishu/constants.py @@ -91,6 +91,12 @@ class FeishuCommandName(StrEnum): RULE_ENABLE = "rule_enable" FINANCE_NEEDS = "finance_needs" PROJECT_FINANCE = "project_finance" + MARKET_OVERVIEW = "market_overview" + MARKET_MACRO = "market_macro" + MARKET_ANNOUNCEMENTS = "market_announcements" + STOCK_ANALYSIS = "stock_analysis" + WATCHLIST_ADD = "watchlist_add" + WATCHLIST_LIST = "watchlist_list" DAILY_BRIEF = "daily_brief" PROJECT_WEEKLY = "project_weekly" ATTENDANCE_SUMMARY = "attendance_summary" diff --git a/app/modules/market/__init__.py b/app/modules/market/__init__.py new file mode 100644 index 0000000..e686845 --- /dev/null +++ b/app/modules/market/__init__.py @@ -0,0 +1,3 @@ +from app.modules.market.service import MarketService + +__all__ = ["MarketService"] diff --git a/app/modules/market/chart.py b/app/modules/market/chart.py new file mode 100644 index 0000000..1afecc2 --- /dev/null +++ b/app/modules/market/chart.py @@ -0,0 +1,62 @@ +from io import BytesIO +from typing import Any + +from PIL import Image, ImageDraw, ImageFont + + +def render_market_chart(report: dict[str, Any]) -> bytes: + image = Image.new("RGB", (1200, 720), (248, 250, 252)) + draw = ImageDraw.Draw(image) + title = _font(34) + label = _font(22) + chart = report.get("chart_data") or {} + if "quotes" in chart: + draw.text( + (55, 35), + f"Stock Analysis: {chart.get('symbol') or ''}", + fill=(17, 24, 39), + font=title, + ) + quotes = chart["quotes"] + values = [float(item["close"]) for item in quotes] + if values: + low, high = min(values), max(values) + span = high - low or 1 + points = [] + for index, value in enumerate(values): + x = 70 + index * 1060 / max(len(values) - 1, 1) + y = 610 - (value - low) * 480 / span + points.append((x, y)) + draw.line(points, fill=(37, 99, 235), width=5) + draw.text( + (70, 630), + f"最低 {low:.2f} 最高 {high:.2f} 最新 {values[-1]:.2f}", + fill=(75, 85, 99), + font=label, + ) + else: + draw.text((55, 35), "A-Share Market Overview", fill=(17, 24, 39), font=title) + values = [ + ("Advances", int(chart.get("advances") or 0)), + ("Declines", int(chart.get("declines") or 0)), + ("Flat", int(chart.get("flat") or 0)), + ] + maximum = max((value for _, value in values), default=1) or 1 + for index, (name, value) in enumerate(values): + y = 170 + index * 150 + draw.text((70, y), f"{name} {value:,}", fill=(75, 85, 99), font=label) + draw.rectangle((70, y + 45, 1120, y + 82), fill=(209, 213, 219)) + draw.rectangle( + (70, y + 45, 70 + 1050 * value / maximum, y + 82), + fill=((37, 99, 235), (220, 38, 38), (245, 158, 11))[index], + ) + output = BytesIO() + image.save(output, format="PNG", optimize=True) + return output.getvalue() + + +def _font(size: int) -> ImageFont.ImageFont: + try: + return ImageFont.truetype("DejaVuSans.ttf", size=size) + except OSError: + return ImageFont.load_default(size=size) diff --git a/app/modules/market/pipeline.py b/app/modules/market/pipeline.py new file mode 100644 index 0000000..a0e0d04 --- /dev/null +++ b/app/modules/market/pipeline.py @@ -0,0 +1,263 @@ +from datetime import date, timedelta +from typing import Any + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.core.config import get_settings +from app.core.constants import ActorValue +from app.core.utils.time import utc_now +from app.modules.feishu.service import FeishuService +from app.modules.market.chart import render_market_chart +from app.modules.market.service import MarketService +from app.modules.reports.constants import ReportPushStatus +from app.modules.reports.services import ReportService +from app.modules.workflows.constants import WorkflowStatus, WorkflowType +from app.modules.workflows.models import WorkflowInstance +from app.modules.workflows.service import WorkflowService + +MARKET_REPORT_TYPES = {"premarket", "close", "weekly"} + + +class MarketPipelineService: + def __init__(self, db: Session, market: MarketService | None = None): + self.db = db + self.market = market or MarketService(db) + self.workflows = WorkflowService(db) + + def period_key(self, report_type: str, reference_date: date) -> str: + self._validate_type(report_type) + period = reference_date + if report_type == "weekly": + period = reference_date - timedelta(days=reference_date.weekday()) + return f"market:{report_type}:{period.isoformat()}" + + def find(self, period_key: str) -> WorkflowInstance | None: + return self.db.execute( + select(WorkflowInstance).where( + WorkflowInstance.workflow_type == WorkflowType.MARKET_ANALYSIS, + WorkflowInstance.aggregate_type == "market_period", + WorkflowInstance.aggregate_id == period_key, + ) + ).scalar_one_or_none() + + def run( + self, + report_type: str, + reference_date: date | None = None, + force: bool = False, + actor: str = ActorValue.SCHEDULER, + ) -> dict[str, Any]: + target = reference_date or date.today() + period_key = self.period_key(report_type, target) + if get_settings().read_only_mode: + return { + "period_key": period_key, + "status": "operations_disabled", + "deduplicated": False, + } + existing = self.find(period_key) + if existing is not None and existing.status == WorkflowStatus.COMPLETED and not force: + return { + "workflow_code": existing.code, + "period_key": period_key, + "status": existing.status, + "deduplicated": True, + } + self._step(period_key, report_type, "source_sync", WorkflowStatus.RUNNING, actor) + try: + sync_result = self._sync(report_type, target) + if sync_result.get("market_closed"): + workflow = self._step( + period_key, + report_type, + "market_closed", + WorkflowStatus.COMPLETED, + actor, + sync_result, + ) + return { + "workflow_code": workflow.code, + "period_key": period_key, + "status": "market_closed", + "deduplicated": False, + } + self._step( + period_key, + report_type, + "ai_analysis", + WorkflowStatus.RUNNING, + actor, + sync_result, + ) + report = ( + self.market.weekly_overview(target, include_ai=True, actor=actor) + if report_type == "weekly" + else self.market.market_overview( + target if report_type == "close" else None, + include_ai=True, + actor=actor, + ) + ) + ai = report.get("ai_analysis") or {} + if not report.get("data_available"): + return self._fail(period_key, report_type, "market_data_unavailable", actor) + if not ai.get("ok"): + self._notify("AI 当前不可用，本次市场分析报告未发送。", actor) + return self._fail(period_key, report_type, "ai_unavailable", actor) + return self._deliver(report_type, period_key, report, force, actor, sync_result) + except Exception as exc: + self.db.rollback() + self._step( + period_key, + report_type, + "failed", + WorkflowStatus.FAILED, + actor, + {"error_type": type(exc).__name__, "error": str(exc)[:1000]}, + ) + self._notify(f"市场分析 {period_key} 执行失败：{type(exc).__name__}", actor) + raise + + def _sync(self, report_type: str, target: date) -> dict[str, Any]: + result: dict[str, Any] = {} + if report_type == "premarket" and not self.market.is_trading_day(target): + return {"market_closed": True} + if report_type == "close": + result["daily"] = self.market.sync_daily(target) + if result["daily"].get("market_closed"): + result["market_closed"] = True + return result + result["macro"] = self.market.sync_macro(target) + start = target - timedelta(days=6 if report_type == "weekly" else 1) + result["announcements"] = self.market.sync_announcements(start, target) + result["financials"] = self.market.sync_watchlist_financials() + return result + + def _deliver( + self, + report_type: str, + period_key: str, + report: dict[str, Any], + force: bool, + actor: str, + sync_result: dict[str, Any], + ) -> dict[str, Any]: + settings = get_settings() + if not ( + settings.feishu_default_chat_id + and settings.feishu_app_id + and settings.feishu_app_secret + ): + return self._fail(period_key, report_type, "delivery_not_configured", actor) + idempotency_key = ( + period_key if not force else f"{period_key}:force:{utc_now():%Y%m%d%H%M%S%f}" + ) + reports = ReportService(self.db) + push_run = reports.create_push_run( + report_type=f"market_{report_type}", + title=report["title"], + receive_id=settings.feishu_default_chat_id, + receive_id_type="chat_id", + actor=actor, + idempotency_key=idempotency_key, + ) + if push_run.status != ReportPushStatus.SUCCESS: + try: + feishu = FeishuService(self.db) + image = feishu.upload_image(render_market_chart(report), actor) + image_key = (image.get("data") or {}).get("image_key") + if not image_key: + raise ValueError("Feishu image upload did not return image_key") + card = FeishuService.build_basic_card( + report["title"], + report["lines"], + image_key=image_key, + image_alt=report["title"], + ) + response = feishu.send_card( + card, + settings.feishu_default_chat_id, + receive_id_type="chat_id", + actor=actor, + ) + reports.update_push_run( + push_run.code, + ReportPushStatus.SUCCESS, + provider_response=response, + sent=True, + ) + except Exception as exc: + reports.update_push_run( + push_run.code, ReportPushStatus.FAILED, error_message=str(exc)[:2000] + ) + raise + workflow = self._step( + period_key, + report_type, + "pushed", + WorkflowStatus.COMPLETED, + actor, + {"push_run_code": push_run.code, "sync": sync_result}, + ) + return { + "workflow_code": workflow.code, + "period_key": period_key, + "push_run_code": push_run.code, + "status": workflow.status, + "deduplicated": False, + } + + def _fail(self, period_key: str, report_type: str, action: str, actor: str) -> dict[str, Any]: + workflow = self._step( + period_key, report_type, action, WorkflowStatus.FAILED, actor + ) + return { + "workflow_code": workflow.code, + "period_key": period_key, + "status": workflow.status, + "reason": action, + "deduplicated": False, + } + + def _step( + self, + period_key: str, + report_type: str, + action: str, + status_value: str, + actor: str, + payload: dict[str, Any] | None = None, + ) -> WorkflowInstance: + return self.workflows.start_or_update( + workflow_type=WorkflowType.MARKET_ANALYSIS, + aggregate_type="market_period", + aggregate_id=period_key, + status_value=status_value, + action=action, + actor=actor, + payload={"report_type": report_type, **(payload or {})}, + ) + + @staticmethod + def _validate_type(report_type: str) -> None: + if report_type not in MARKET_REPORT_TYPES: + raise ValueError("Market report type must be premarket, close or weekly") + + def _notify(self, content: str, actor: str) -> None: + settings = get_settings() + if not ( + settings.feishu_default_chat_id + and settings.feishu_app_id + and settings.feishu_app_secret + ): + return + try: + FeishuService(self.db).send_text( + content, + settings.feishu_default_chat_id, + receive_id_type="chat_id", + actor=actor, + ) + except Exception: + self.db.rollback() diff --git a/app/modules/market/routes.py b/app/modules/market/routes.py new file mode 100644 index 0000000..3fccbcf --- /dev/null +++ b/app/modules/market/routes.py @@ -0,0 +1,130 @@ +from datetime import date +from typing import Literal + +from fastapi import APIRouter, Depends +from pydantic import BaseModel +from sqlalchemy.orm import Session + +from app.core.database import get_db +from app.core.background.task_queue.market import enqueue_market_report +from app.core.security import ApiPrincipal, require_api_key, require_operations_enabled +from app.modules.market.service import MarketService + +router = APIRouter(dependencies=[Depends(require_api_key)]) + + +class WatchlistRequest(BaseModel): + symbol: str + + +class MarketReportRequest(BaseModel): + report_type: Literal["premarket", "close", "weekly"] + reference_date: date | None = None + force: bool = False + + +@router.get("/overview") +def overview( + trade_date: date | None = None, + include_ai: bool = False, + db: Session = Depends(get_db), + principal: ApiPrincipal = Depends(require_api_key), +) -> dict: + return MarketService(db).market_overview(trade_date, include_ai, principal.actor) + + +@router.get("/stocks/{symbol}/analysis") +def stock_analysis( + symbol: str, + include_ai: bool = False, + db: Session = Depends(get_db), + principal: ApiPrincipal = Depends(require_api_key), +) -> dict: + return MarketService(db).stock_analysis(symbol, include_ai, principal.actor) + + +@router.get("/weekly") +def weekly_overview( + reference_date: date | None = None, + include_ai: bool = False, + db: Session = Depends(get_db), + principal: ApiPrincipal = Depends(require_api_key), +) -> dict: + return MarketService(db).weekly_overview(reference_date, include_ai, principal.actor) + + +@router.get("/industries/{industry}/analysis") +def industry_analysis(industry: str, db: Session = Depends(get_db)) -> dict: + return MarketService(db).industry_analysis(industry) + + +@router.get("/stocks/compare/{first}/{second}") +def compare_stocks(first: str, second: str, db: Session = Depends(get_db)) -> dict: + return MarketService(db).compare_stocks([first, second]) + + +@router.get("/macro") +def macro_overview( + include_ai: bool = False, + db: Session = Depends(get_db), + principal: ApiPrincipal = Depends(require_api_key), +) -> dict: + service = MarketService(db) + return ( + service.macro_analysis(include_ai=True, actor=principal.actor) + if include_ai + else service.macro_overview() + ) + + +@router.get("/announcements") +def announcements( + symbol: str | None = None, + start_date: date | None = None, + limit: int = 50, + db: Session = Depends(get_db), +) -> dict: + return MarketService(db).announcements(symbol, start_date, limit) + + +@router.post("/sync/daily") +def sync_daily(trade_date: date, db: Session = Depends(get_db)) -> dict: + require_operations_enabled() + return MarketService(db).sync_daily(trade_date) + + +@router.post("/sync/macro") +def sync_macro(reference_date: date | None = None, db: Session = Depends(get_db)) -> dict: + require_operations_enabled() + return MarketService(db).sync_macro(reference_date) + + +@router.post("/sync/announcements") +def sync_announcements( + start_date: date, end_date: date, db: Session = Depends(get_db) +) -> dict: + require_operations_enabled() + return {"processed": MarketService(db).sync_announcements(start_date, end_date)} + + +@router.post("/reports/enqueue") +def enqueue_report(payload: MarketReportRequest) -> dict: + require_operations_enabled() + return enqueue_market_report(payload.report_type, payload.reference_date, payload.force) + + +@router.post("/watchlist") +def add_watchlist( + payload: WatchlistRequest, + db: Session = Depends(get_db), + principal: ApiPrincipal = Depends(require_api_key), +) -> dict: + require_operations_enabled() + return MarketService(db).add_watchlist(principal.actor, payload.symbol) + + +@router.get("/watchlist") +def watchlist( + db: Session = Depends(get_db), principal: ApiPrincipal = Depends(require_api_key) +) -> dict: + return {"items": MarketService(db).watchlist(principal.actor)} diff --git a/app/modules/market/service.py b/app/modules/market/service.py new file mode 100644 index 0000000..d47717c --- /dev/null +++ b/app/modules/market/service.py @@ -0,0 +1,1043 @@ +import math +from calendar import monthrange +from datetime import date, datetime, timedelta +from decimal import Decimal, InvalidOperation +from hashlib import sha256 +from statistics import pstdev +from typing import Any + +import httpx +from fastapi import HTTPException +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.core.config import get_settings +from app.core.utils.time import utc_now +from app.modules.ai_agent.constants import AIProviderName, AIResponseKey +from app.modules.ai_agent.service import AIService +from app.modules.ai_agent.skills import AISkillId +from app.modules.audit.constants import AuditAction, AuditRiskLevel, AuditSource, AuditTargetType +from app.modules.audit.schemas import AuditLogCreate +from app.modules.audit.service import AuditService +from app.modules.business.models import ( + MarketAnnouncement, + MarketDailyQuote, + MarketFinancialMetric, + MarketInstrument, + MarketMacroIndicator, + MarketWatchlist, +) + +ALLOWED_APIS = frozenset( + { + "stock_basic", + "trade_cal", + "daily", + "daily_basic", + "index_daily", + "fina_indicator", + "anns_d", + "shibor", + "cn_cpi", + "cn_gdp", + "cn_m", + } +) +INDEX_SYMBOLS = { + "000001.SH": "上证指数", + "399001.SZ": "深证成指", + "399006.SZ": "创业板指", + "000688.SH": "科创50", +} + + +class TushareClient: + def __init__(self) -> None: + settings = get_settings() + if settings.market_data_provider != "tushare": + raise HTTPException(status_code=503, detail="Unsupported market data provider") + if not settings.market_data_token: + raise HTTPException(status_code=503, detail="MARKET_DATA_TOKEN is not configured") + self.base_url = settings.market_data_base_url + self.token = settings.market_data_token + + def query(self, api_name: str, params: dict[str, Any], fields: list[str]) -> list[dict]: + if api_name not in ALLOWED_APIS: + raise HTTPException(status_code=403, detail="Market API is not allowed") + try: + response = httpx.post( + self.base_url, + json={ + "api_name": api_name, + "token": self.token, + "params": params, + "fields": ",".join(fields), + }, + timeout=30, + ) + response.raise_for_status() + payload = response.json() + except (httpx.HTTPError, ValueError) as exc: + raise HTTPException(status_code=503, detail="Market data provider unavailable") from exc + if payload.get("code") != 0: + raise HTTPException( + status_code=502, detail=str(payload.get("msg") or "Market provider error") + ) + data = payload.get("data") or {} + names = data.get("fields") or [] + return [dict(zip(names, item, strict=False)) for item in data.get("items") or []] + + +class MarketService: + def __init__(self, db: Session, provider: TushareClient | None = None): + self.db = db + self.provider = provider + + def sync_daily(self, trade_date: date) -> dict[str, int]: + provider = self.provider or TushareClient() + raw_date = trade_date.strftime("%Y%m%d") + if not self.is_trading_day(trade_date, provider): + result = {"instruments": 0, "quotes": 0, "valuations": 0, "market_closed": 1} + self._audit_sync("daily", result) + return result + instruments = provider.query( + "stock_basic", + {"list_status": "L"}, + ["ts_code", "symbol", "name", "area", "industry", "market", "exchange", "list_date"], + ) + self._upsert_instruments(instruments) + quotes = provider.query( + "daily", + {"trade_date": raw_date}, + [ + "ts_code", + "trade_date", + "open", + "high", + "low", + "close", + "pre_close", + "pct_chg", + "vol", + "amount", + ], + ) + valuations = provider.query( + "daily_basic", + {"trade_date": raw_date}, + ["ts_code", "trade_date", "pe_ttm", "pb", "total_mv"], + ) + valuation_map = {row["ts_code"]: row for row in valuations} + for symbol in INDEX_SYMBOLS: + quotes.extend( + provider.query( + "index_daily", + {"ts_code": symbol, "trade_date": raw_date}, + [ + "ts_code", + "trade_date", + "open", + "high", + "low", + "close", + "pre_close", + "pct_chg", + "vol", + "amount", + ], + ) + ) + self._ensure_indices() + self._upsert_quotes(quotes, valuation_map) + result = { + "instruments": len(instruments), + "quotes": len(quotes), + "valuations": len(valuations), + } + self._audit_sync("daily", result) + return result + + def is_trading_day( + self, trade_date: date, provider: TushareClient | None = None + ) -> bool: + source = provider or self.provider or TushareClient() + raw_date = trade_date.strftime("%Y%m%d") + calendar = source.query( + "trade_cal", + {"exchange": "SSE", "start_date": raw_date, "end_date": raw_date}, + ["exchange", "cal_date", "is_open", "pretrade_date"], + ) + return bool(calendar and str(calendar[0].get("is_open")) == "1") + + def sync_financials(self, symbol: str) -> int: + provider = self.provider or TushareClient() + code = normalize_symbol(symbol) + target = date.today() + raw_rows = provider.query( + "fina_indicator", + { + "ts_code": code, + "start_date": date(target.year - 3, 1, 1).strftime("%Y%m%d"), + "end_date": target.strftime("%Y%m%d"), + }, + [ + "ts_code", + "ann_date", + "end_date", + "update_flag", + "or_yoy", + "netprofit_yoy", + "roe", + "debt_to_assets", + "ocf_yoy", + ], + ) + rows: dict[date, dict[str, Any]] = {} + for row in raw_rows: + period = _date(row.get("end_date")) + if not period: + continue + current = rows.get(period) + if current is None or _financial_row_order(row) > _financial_row_order(current): + rows[period] = row + for period, row in rows.items(): + record = self.db.execute( + select(MarketFinancialMetric).where( + MarketFinancialMetric.symbol == code, MarketFinancialMetric.period_end == period + ) + ).scalar_one_or_none() + values = { + "revenue_yoy": _decimal(row.get("or_yoy")), + "net_profit_yoy": _decimal(row.get("netprofit_yoy")), + "roe": _decimal(row.get("roe")), + "debt_to_assets": _decimal(row.get("debt_to_assets")), + "operating_cashflow_yoy": _decimal(row.get("ocf_yoy")), + } + if record is None: + self.db.add(MarketFinancialMetric(symbol=code, period_end=period, **values)) + else: + for key, value in values.items(): + setattr(record, key, value) + self.db.commit() + self._audit_sync( + "financials", {"symbol": code, "raw": len(raw_rows), "processed": len(rows)} + ) + return len(rows) + + def sync_macro(self, reference_date: date | None = None) -> dict[str, int]: + provider = self.provider or TushareClient() + target = reference_date or date.today() + start = target - timedelta(days=400) + month_params = {"start_m": start.strftime("%Y%m"), "end_m": target.strftime("%Y%m")} + quarter_params = { + "start_q": f"{start.year}Q1", + "end_q": f"{target.year}Q{(target.month - 1) // 3 + 1}", + } + sources = ( + ( + "shibor", + {"start_date": start.strftime("%Y%m%d"), "end_date": target.strftime("%Y%m%d")}, + ["date", "on"], + "date", + (("SHIBOR_ON", "隔夜Shibor", "on", "%"),), + ), + ( + "cn_cpi", + month_params, + ["month", "nt_yoy"], + "month", + (("CPI_YOY", "全国CPI同比", "nt_yoy", "%"),), + ), + ( + "cn_gdp", + quarter_params, + ["quarter", "gdp_yoy"], + "quarter", + (("GDP_YOY", "国内生产总值同比", "gdp_yoy", "%"),), + ), + ( + "cn_m", + month_params, + ["month", "m2_yoy"], + "month", + (("M2_YOY", "M2同比", "m2_yoy", "%"),), + ), + ) + result: dict[str, int] = {} + for api_name, params, fields, period_field, indicators in sources: + rows = provider.query(api_name, params, fields) + written = 0 + for row in rows: + period = _macro_period(row.get(period_field), period_field) + if period is None: + continue + for code, name, value_field, unit in indicators: + value = _decimal(row.get(value_field)) + if value is None: + continue + record = self.db.execute( + select(MarketMacroIndicator).where( + MarketMacroIndicator.code == code, + MarketMacroIndicator.period_date == period, + ) + ).scalar_one_or_none() + values = {"name": name, "value": value, "unit": unit, "source": "tushare"} + if record is None: + self.db.add( + MarketMacroIndicator(code=code, period_date=period, **values) + ) + else: + for key, item in values.items(): + setattr(record, key, item) + written += 1 + result[api_name] = written + self.db.commit() + self._audit_sync("macro", result) + return result + + def sync_announcements(self, start_date: date, end_date: date) -> int: + if end_date < start_date or (end_date - start_date).days > 30: + raise HTTPException(status_code=422, detail="Announcement range must be 1 to 31 days") + provider = self.provider or TushareClient() + rows: list[dict] = [] + current = start_date + while current <= end_date: + rows.extend( + provider.query( + "anns_d", + {"ann_date": current.strftime("%Y%m%d")}, + ["ann_date", "ts_code", "name", "title", "url", "rec_time"], + ) + ) + current += timedelta(days=1) + written = 0 + for row in rows: + announced = _date(row.get("ann_date")) + title = str(row.get("title") or "").strip() + if announced is None or not title: + continue + symbol = str(row.get("ts_code") or "").strip() or None + source_url = str(row.get("url") or "").strip() or None + source_key = sha256( + f"{symbol or ''}|{announced.isoformat()}|{title}|{source_url or ''}".encode() + ).hexdigest() + record = self.db.execute( + select(MarketAnnouncement).where(MarketAnnouncement.source_key == source_key) + ).scalar_one_or_none() + values = { + "symbol": symbol, + "announcement_date": announced, + "published_at": _datetime(row.get("rec_time")), + "title": title[:512], + "category": None, + "source_url": source_url, + "source": "tushare", + } + if record is None: + self.db.add(MarketAnnouncement(source_key=source_key, **values)) + else: + for key, value in values.items(): + setattr(record, key, value) + written += 1 + self.db.commit() + self._audit_sync("announcements", {"processed": written}) + return written + + def sync_watchlist_financials(self) -> dict[str, int]: + symbols = set( + self.db.execute( + select(MarketWatchlist.symbol).where(MarketWatchlist.enabled.is_(True)) + ).scalars() + ) + return {symbol: self.sync_financials(symbol) for symbol in sorted(symbols)} + + def announcements( + self, symbol: str | None = None, start_date: date | None = None, limit: int = 50 + ) -> dict[str, Any]: + stmt = select(MarketAnnouncement).order_by( + MarketAnnouncement.announcement_date.desc(), MarketAnnouncement.id.desc() + ) + if symbol: + stmt = stmt.where(MarketAnnouncement.symbol == normalize_symbol(symbol)) + if start_date: + stmt = stmt.where(MarketAnnouncement.announcement_date >= start_date) + rows = list(self.db.execute(stmt.limit(min(max(limit, 1), 100))).scalars()) + return { + "items": [ + { + "symbol": row.symbol, + "announcement_date": row.announcement_date.isoformat(), + "published_at": row.published_at.isoformat() if row.published_at else None, + "title": row.title, + "category": row.category, + "source_url": row.source_url, + "source": row.source, + } + for row in rows + ] + } + + def market_overview( + self, trade_date: date | None = None, include_ai: bool = False, actor: str = "api" + ) -> dict[str, Any]: + target = ( + trade_date + or self.db.execute( + select(MarketDailyQuote.trade_date) + .order_by(MarketDailyQuote.trade_date.desc()) + .limit(1) + ).scalar() + ) + if target is None: + return { + "title": "A股市场分析", + "data_available": False, + "content": "市场数据未接入。", + "lines": ["- 市场数据未接入。"], + "ai_analysis": None, + } + rows = list( + self.db.execute( + select(MarketDailyQuote, MarketInstrument) + .join(MarketInstrument, MarketInstrument.symbol == MarketDailyQuote.symbol) + .where(MarketDailyQuote.trade_date == target) + ).all() + ) + if not rows: + return { + "title": "A股市场分析", + "data_available": False, + "content": f"{target.isoformat()} 没有可用市场数据。", + "lines": [f"- {target.isoformat()} 没有可用市场数据。"], + "ai_analysis": None, + } + stocks = [(q, i) for q, i in rows if i.instrument_type == "stock"] + indices = [ + { + "symbol": i.symbol, + "name": i.name, + "close": float(q.close_price), + "pct_change": _float(q.pct_change), + } + for q, i in rows + if i.instrument_type == "index" + ] + advances = sum(1 for q, _ in stocks if _float(q.pct_change) > 0) + declines = sum(1 for q, _ in stocks if _float(q.pct_change) < 0) + sectors: dict[str, list[float]] = {} + for quote, instrument in stocks: + sectors.setdefault(instrument.industry or "未分类", []).append(_float(quote.pct_change)) + sector_items = sorted( + ( + { + "industry": name, + "average_pct_change": round(sum(values) / len(values), 2), + "stocks": len(values), + } + for name, values in sectors.items() + ), + key=lambda item: item["average_pct_change"], + reverse=True, + ) + metrics = { + "trade_date": target.isoformat(), + "stocks": len(stocks), + "advances": advances, + "declines": declines, + "flat": len(stocks) - advances - declines, + "limit_up": sum(1 for q, _ in stocks if _float(q.pct_change) >= 9.5), + "limit_down": sum(1 for q, _ in stocks if _float(q.pct_change) <= -9.5), + "turnover_cny": round(sum(_float(q.amount_cny) for q, _ in stocks), 2), + "indices": indices, + "top_sectors": sector_items[:5], + "bottom_sectors": sector_items[-5:], + "macro": self.macro_overview()["items"][:10], + "announcements": _ai_announcements( + self.announcements(start_date=target - timedelta(days=3), limit=10)["items"] + ), + } + lines = [ + f"- 交易日：{target.isoformat()}，上涨 {advances}，下跌 {declines}，平盘 {metrics['flat']}", + f"- 涨停/跌停：{metrics['limit_up']}/{metrics['limit_down']}，成交额 ¥{metrics['turnover_cny']:,.0f}", + "- 领涨行业：" + + "、".join(f"{x['industry']} {x['average_pct_change']}%" for x in sector_items[:5]), + ] + report = { + "title": "A股市场分析", + "data_available": True, + "metrics": metrics, + "chart_data": metrics, + "lines": lines, + "content": "\n".join(lines), + } + report["ai_analysis"] = ( + self._ai(AISkillId.MARKET_OVERVIEW_ANALYSIS, report, actor) if include_ai else None + ) + if report["ai_analysis"] and report["ai_analysis"].get("ok"): + lines.append("- AI 市场分析：") + lines.extend(f" {line}" for line in report["ai_analysis"]["answer"].splitlines()) + report["content"] = "\n".join(lines) + return report + + def weekly_overview( + self, reference_date: date | None = None, include_ai: bool = False, actor: str = "api" + ) -> dict[str, Any]: + end_date = reference_date or date.today() + trade_dates = list( + self.db.execute( + select(MarketDailyQuote.trade_date) + .where(MarketDailyQuote.trade_date <= end_date) + .distinct() + .order_by(MarketDailyQuote.trade_date.desc()) + .limit(5) + ).scalars() + ) + if not trade_dates: + return { + "title": "A股市场周报", + "data_available": False, + "content": "市场数据未接入。", + "lines": ["- 市场数据未接入。"], + "ai_analysis": None, + } + first_date, last_date = min(trade_dates), max(trade_dates) + rows = list( + self.db.execute( + select(MarketDailyQuote, MarketInstrument) + .join(MarketInstrument, MarketInstrument.symbol == MarketDailyQuote.symbol) + .where(MarketDailyQuote.trade_date.in_({first_date, last_date})) + ).all() + ) + by_symbol: dict[str, dict[date, tuple[MarketDailyQuote, MarketInstrument]]] = {} + for quote, instrument in rows: + by_symbol.setdefault(quote.symbol, {})[quote.trade_date] = (quote, instrument) + stocks: list[tuple[float, MarketInstrument]] = [] + indices: list[dict[str, Any]] = [] + for values in by_symbol.values(): + if first_date not in values or last_date not in values: + continue + first_quote, instrument = values[first_date] + last_quote, _ = values[last_date] + if not first_quote.close_price: + continue + change = round((float(last_quote.close_price) / float(first_quote.close_price) - 1) * 100, 2) + if instrument.instrument_type == "stock": + stocks.append((change, instrument)) + elif instrument.instrument_type == "index": + indices.append( + {"symbol": instrument.symbol, "name": instrument.name, "pct_change": change} + ) + if not stocks and not indices: + return { + "title": "A股市场周报", + "data_available": False, + "content": "最近交易日缺少可比较的市场数据。", + "lines": ["- 最近交易日缺少可比较的市场数据。"], + "ai_analysis": None, + } + sectors: dict[str, list[float]] = {} + for change, instrument in stocks: + sectors.setdefault(instrument.industry or "未分类", []).append(change) + sector_items = sorted( + ( + { + "industry": name, + "average_pct_change": round(sum(values) / len(values), 2), + "stocks": len(values), + } + for name, values in sectors.items() + ), + key=lambda item: item["average_pct_change"], + reverse=True, + ) + advances = sum(1 for value, _ in stocks if value > 0) + declines = sum(1 for value, _ in stocks if value < 0) + metrics = { + "period_start": first_date.isoformat(), + "period_end": last_date.isoformat(), + "trading_days": len(trade_dates), + "stocks": len(stocks), + "advances": advances, + "declines": declines, + "flat": len(stocks) - advances - declines, + "indices": sorted(indices, key=lambda item: item["symbol"]), + "top_sectors": sector_items[:5], + "bottom_sectors": sector_items[-5:], + "macro": self.macro_overview()["items"][:10], + "announcements": _ai_announcements( + self.announcements(start_date=first_date, limit=10)["items"] + ), + } + lines = [ + f"- 周期：{first_date.isoformat()} 至 {last_date.isoformat()}，共 {len(trade_dates)} 个交易日", + f"- 区间上涨 {advances}，下跌 {declines}，平盘 {metrics['flat']}", + "- 领涨行业：" + + ("、".join(f"{x['industry']} {x['average_pct_change']}%" for x in sector_items[:5]) or "无"), + ] + report = { + "title": "A股市场周报", + "data_available": True, + "metrics": metrics, + "chart_data": metrics, + "lines": lines, + "content": "\n".join(lines), + } + report["ai_analysis"] = ( + self._ai(AISkillId.MARKET_OVERVIEW_ANALYSIS, report, actor) if include_ai else None + ) + if report["ai_analysis"] and report["ai_analysis"].get("ok"): + lines.append("- AI 市场周度分析：") + lines.extend(f" {line}" for line in report["ai_analysis"]["answer"].splitlines()) + report["content"] = "\n".join(lines) + return report + + def stock_analysis( + self, symbol: str, include_ai: bool = False, actor: str = "api" + ) -> dict[str, Any]: + code = normalize_symbol(symbol) + instrument = self.db.execute( + select(MarketInstrument).where(MarketInstrument.symbol == code) + ).scalar_one_or_none() + quotes = list( + self.db.execute( + select(MarketDailyQuote) + .where(MarketDailyQuote.symbol == code) + .order_by(MarketDailyQuote.trade_date.desc()) + .limit(120) + ).scalars() + ) + if instrument is None or not quotes: + raise HTTPException(status_code=404, detail="Stock market data not found") + quotes.reverse() + closes = [float(q.close_price) for q in quotes] + latest = quotes[-1] + returns = [(closes[i] / closes[i - 1] - 1) for i in range(1, len(closes)) if closes[i - 1]] + financial = self.db.execute( + select(MarketFinancialMetric) + .where(MarketFinancialMetric.symbol == code) + .order_by(MarketFinancialMetric.period_end.desc()) + .limit(1) + ).scalar_one_or_none() + metrics = { + "symbol": code, + "name": instrument.name, + "trade_date": latest.trade_date.isoformat(), + "close": float(latest.close_price), + "pct_change": _float(latest.pct_change), + "return_5d": _period_return(closes, 5), + "return_20d": _period_return(closes, 20), + "return_60d": _period_return(closes, 60), + "ma5": _mean(closes[-5:]), + "ma20": _mean(closes[-20:]), + "ma60": _mean(closes[-60:]), + "annualized_volatility": round(pstdev(returns) * math.sqrt(250) * 100, 2) + if len(returns) > 1 + else None, + "max_drawdown": _max_drawdown(closes), + "rsi14": _rsi(closes, 14), + "pe": _float(latest.pe) if latest.pe is not None else None, + "pb": _float(latest.pb) if latest.pb is not None else None, + "financial": _financial(financial), + } + lines = [ + f"- {instrument.name}（{code}）收盘 {metrics['close']}，当日 {metrics['pct_change']}%", + f"- 5/20/60日收益：{metrics['return_5d']}% / {metrics['return_20d']}% / {metrics['return_60d']}%", + f"- 波动率 {metrics['annualized_volatility']}%，最大回撤 {metrics['max_drawdown']}%，RSI14 {metrics['rsi14']}", + f"- PE/PB：{metrics['pe']} / {metrics['pb']}", + ] + report = { + "title": f"股票分析：{instrument.name}", + "data_available": True, + "metrics": metrics, + "chart_data": { + "symbol": code, + "name": instrument.name, + "quotes": [ + {"date": q.trade_date.isoformat(), "close": float(q.close_price)} + for q in quotes[-60:] + ], + }, + "lines": lines, + "content": "\n".join(lines), + } + report["ai_analysis"] = ( + self._ai(AISkillId.STOCK_ANALYSIS, report, actor) if include_ai else None + ) + if report["ai_analysis"] and report["ai_analysis"].get("ok"): + lines.append("- AI 股票研究：") + lines.extend(f" {line}" for line in report["ai_analysis"]["answer"].splitlines()) + report["content"] = "\n".join(lines) + return report + + def industry_analysis(self, industry: str) -> dict[str, Any]: + latest = self.db.execute( + select(MarketDailyQuote.trade_date) + .order_by(MarketDailyQuote.trade_date.desc()) + .limit(1) + ).scalar() + if latest is None: + raise HTTPException(status_code=404, detail="Market data not found") + rows = list( + self.db.execute( + select(MarketDailyQuote, MarketInstrument) + .join(MarketInstrument, MarketInstrument.symbol == MarketDailyQuote.symbol) + .where( + MarketDailyQuote.trade_date == latest, + MarketInstrument.industry == industry, + ) + ).all() + ) + if not rows: + raise HTTPException(status_code=404, detail="Industry data not found") + items = sorted( + [ + { + "symbol": instrument.symbol, + "name": instrument.name, + "pct_change": _float(quote.pct_change), + "pe": _float(quote.pe) if quote.pe is not None else None, + } + for quote, instrument in rows + ], + key=lambda item: item["pct_change"], + reverse=True, + ) + return { + "title": f"行业分析：{industry}", + "trade_date": latest.isoformat(), + "industry": industry, + "average_pct_change": round(sum(item["pct_change"] for item in items) / len(items), 2), + "items": items, + } + + def compare_stocks(self, symbols: list[str]) -> dict[str, Any]: + if len(symbols) != 2: + raise HTTPException(status_code=422, detail="Exactly two stock symbols are required") + reports = [self.stock_analysis(symbol) for symbol in symbols] + return { + "title": "股票对比", + "items": [report["metrics"] for report in reports], + "content": "\n".join( + f"- {item['name']}（{item['symbol']}）：20日收益 {item['return_20d']}%，" + f"波动率 {item['annualized_volatility']}%，PE/PB {item['pe']}/{item['pb']}" + for item in (report["metrics"] for report in reports) + ), + } + + def macro_overview(self) -> dict[str, Any]: + rows = list( + self.db.execute( + select(MarketMacroIndicator).order_by( + MarketMacroIndicator.period_date.desc(), MarketMacroIndicator.code + ) + ).scalars() + ) + latest: dict[str, MarketMacroIndicator] = {} + for row in rows: + latest.setdefault(row.code, row) + items = [ + { + "code": item.code, + "name": item.name, + "period_date": item.period_date.isoformat(), + "value": float(item.value), + "unit": item.unit, + "source": item.source, + } + for item in latest.values() + ] + return { + "title": "金融市场宏观分析", + "data_available": bool(items), + "items": items, + "content": ( + "宏观金融数据未接入。" + if not items + else "\n".join( + f"- {item['name']}：{item['value']} {item['unit'] or ''}" + f"（{item['period_date']}）" + for item in items + ) + ), + } + + def macro_analysis(self, include_ai: bool = False, actor: str = "api") -> dict[str, Any]: + overview = self.macro_overview() + lines = overview["content"].splitlines() + report = { + **overview, + "metrics": {"macro": overview["items"]}, + "lines": lines, + } + report["ai_analysis"] = ( + self._ai(AISkillId.MARKET_OVERVIEW_ANALYSIS, report, actor) + if include_ai and overview["data_available"] + else None + ) + if report["ai_analysis"] and report["ai_analysis"].get("ok"): + lines.append("- AI 宏观金融分析：") + lines.extend(f" {line}" for line in report["ai_analysis"]["answer"].splitlines()) + report["content"] = "\n".join(lines) + return report + + def add_watchlist(self, actor: str, symbol: str) -> dict[str, Any]: + code = normalize_symbol(symbol) + record = self.db.execute( + select(MarketWatchlist).where( + MarketWatchlist.actor == actor, MarketWatchlist.symbol == code + ) + ).scalar_one_or_none() + if record is None: + record = MarketWatchlist(actor=actor, symbol=code) + self.db.add(record) + else: + record.enabled = True + self.db.commit() + AuditService(self.db).log( + AuditLogCreate( + actor=actor, + source=AuditSource.MARKET, + action=AuditAction.MARKET_WATCHLIST_UPDATE, + target_type=AuditTargetType.MARKET, + target_id=code, + risk_level=AuditRiskLevel.LOW, + response_payload={"enabled": True}, + ) + ) + return {"actor": actor, "symbol": code, "enabled": True} + + def watchlist(self, actor: str) -> list[dict[str, Any]]: + records = self.db.execute( + select(MarketWatchlist).where( + MarketWatchlist.actor == actor, MarketWatchlist.enabled.is_(True) + ) + ).scalars() + return [{"symbol": r.symbol} for r in records] + + def _ai(self, skill: AISkillId, report: dict[str, Any], actor: str) -> dict[str, Any]: + try: + result = AIService(self.db).run_skill( + skill, + context={ + "metrics": report["metrics"], + "memory_scope": "market", + "memory_subject": "market", + "scope": "market", + "subject": "market", + }, + actor=actor, + ) + except Exception as exc: + self.db.rollback() + return {"ok": False, "error": str(exc), "type": type(exc).__name__} + if result.get(AIResponseKey.PROVIDER) == AIProviderName.NOOP: + return {"ok": False, "error": "AI unavailable"} + return {"ok": True, **result} + + def _upsert_instruments(self, rows: list[dict]) -> None: + existing = {r.symbol: r for r in self.db.execute(select(MarketInstrument)).scalars()} + for row in rows: + code = row["ts_code"] + record = existing.get(code) + values = { + "name": row.get("name") or code, + "exchange": row.get("exchange"), + "instrument_type": "stock", + "industry": row.get("industry"), + "list_date": _date(row.get("list_date")), + "is_active": True, + "source_updated_at": utc_now(), + } + if record is None: + self.db.add(MarketInstrument(symbol=code, **values)) + else: + for key, value in values.items(): + setattr(record, key, value) + self.db.commit() + + def _audit_sync(self, dataset: str, result: dict[str, Any]) -> None: + AuditService(self.db).log( + AuditLogCreate( + actor="market-sync", + source=AuditSource.MARKET, + action=AuditAction.MARKET_SYNC, + target_type=AuditTargetType.MARKET, + target_id=dataset, + risk_level=AuditRiskLevel.LOW, + response_payload=result, + ) + ) + + def _ensure_indices(self) -> None: + existing = { + r.symbol + for r in self.db.execute( + select(MarketInstrument).where(MarketInstrument.symbol.in_(INDEX_SYMBOLS)) + ).scalars() + } + for symbol, name in INDEX_SYMBOLS.items(): + if symbol not in existing: + self.db.add( + MarketInstrument( + symbol=symbol, + name=name, + exchange=symbol.split(".")[-1], + instrument_type="index", + ) + ) + self.db.commit() + + def _upsert_quotes(self, rows: list[dict], valuations: dict[str, dict]) -> None: + for row in rows: + day = _date(row.get("trade_date")) + code = row.get("ts_code") + if not day or not code or row.get("close") is None: + continue + record = self.db.execute( + select(MarketDailyQuote).where( + MarketDailyQuote.symbol == code, MarketDailyQuote.trade_date == day + ) + ).scalar_one_or_none() + basic = valuations.get(code, {}) + values = { + "open_price": _decimal(row.get("open")), + "high_price": _decimal(row.get("high")), + "low_price": _decimal(row.get("low")), + "close_price": _decimal(row.get("close")), + "pre_close": _decimal(row.get("pre_close")), + "pct_change": _decimal(row.get("pct_chg")), + "volume": _decimal(row.get("vol")), + "amount_cny": (_decimal(row.get("amount")) or Decimal(0)) * 1000, + "pe": _decimal(basic.get("pe_ttm")), + "pb": _decimal(basic.get("pb")), + "total_market_value": (_decimal(basic.get("total_mv")) or Decimal(0)) * 10000, + } + if record is None: + self.db.add(MarketDailyQuote(symbol=code, trade_date=day, **values)) + else: + for key, value in values.items(): + setattr(record, key, value) + self.db.commit() + + +def normalize_symbol(value: str) -> str: + text = value.strip().upper() + if "." in text: + return text + if not text.isdigit() or len(text) != 6: + raise HTTPException(status_code=422, detail="Invalid A-share symbol") + return f"{text}.{'SH' if text[0] in {'6','9'} else 'BJ' if text[0] in {'4','8'} else 'SZ'}" + + +def _decimal(value: Any) -> Decimal | None: + try: + return None if value in (None, "") else Decimal(str(value)) + except (InvalidOperation, ValueError): + return None + + +def _float(value: Any) -> float: + return round(float(value or 0), 4) + + +def _date(value: Any) -> date | None: + if isinstance(value, date): + return value + try: + return datetime.strptime(str(value), "%Y%m%d").date() + except (TypeError, ValueError): + return None + + +def _datetime(value: Any) -> datetime | None: + if isinstance(value, datetime): + return value + text = str(value or "").strip() + for pattern in ("%Y-%m-%d %H:%M:%S", "%Y%m%d%H%M%S", "%Y%m%d %H:%M:%S"): + try: + return datetime.strptime(text, pattern) + except ValueError: + continue + return None + + +def _macro_period(value: Any, kind: str) -> date | None: + text = str(value or "").strip().upper() + try: + if kind == "date": + return datetime.strptime(text, "%Y%m%d").date() + if kind == "month" and len(text) == 6: + year, month = int(text[:4]), int(text[4:]) + return date(year, month, monthrange(year, month)[1]) + if kind == "quarter" and len(text) == 6 and text[4] == "Q": + year, quarter = int(text[:4]), int(text[5]) + month = quarter * 3 + return date(year, month, monthrange(year, month)[1]) + except ValueError: + return None + return None + + +def _mean(values: list[float]) -> float | None: + return round(sum(values) / len(values), 4) if values else None + + +def _period_return(values: list[float], days: int) -> float | None: + return ( + round((values[-1] / values[-days - 1] - 1) * 100, 2) + if len(values) > days and values[-days - 1] + else None + ) + + +def _max_drawdown(values: list[float]) -> float | None: + if not values: + return None + peak = values[0] + worst = 0.0 + for value in values: + peak = max(peak, value) + worst = min(worst, value / peak - 1) + return round(worst * 100, 2) + + +def _rsi(values: list[float], days: int) -> float | None: + if len(values) <= days: + return None + changes = [values[i] - values[i - 1] for i in range(len(values) - days, len(values))] + gains = sum(max(x, 0) for x in changes) / days + losses = sum(max(-x, 0) for x in changes) / days + return 100.0 if not losses else round(100 - 100 / (1 + gains / losses), 2) + + +def _financial(record: MarketFinancialMetric | None) -> dict[str, Any] | None: + if record is None: + return None + return { + "period_end": record.period_end.isoformat(), + "revenue_yoy": _optional_float(record.revenue_yoy), + "net_profit_yoy": _optional_float(record.net_profit_yoy), + "roe": _optional_float(record.roe), + "debt_to_assets": _optional_float(record.debt_to_assets), + "operating_cashflow_yoy": _optional_float(record.operating_cashflow_yoy), + } + + +def _optional_float(value: Any) -> float | None: + return None if value is None else round(float(value), 4) + + +def _financial_row_order(row: dict[str, Any]) -> tuple[bool, str]: + return str(row.get("update_flag") or "") == "1", str(row.get("ann_date") or "") + + +def _ai_announcements(items: list[dict[str, Any]]) -> list[dict[str, Any]]: + return [ + { + "symbol": item["symbol"], + "announcement_date": item["announcement_date"], + "title": item["title"], + "category": item["category"], + } + for item in items + ] diff --git a/app/modules/reports/lifecycle_pipeline.py b/app/modules/reports/lifecycle_pipeline.py index 053b9a4..c8154f9 100644 --- a/app/modules/reports/lifecycle_pipeline.py +++ b/app/modules/reports/lifecycle_pipeline.py @@ -64,6 +64,12 @@ class LifecyclePipelineService: force: bool = False, actor: str = ActorValue.SCHEDULER, ) -> dict[str, Any]: + if get_settings().read_only_mode: + return { + "period_key": self.period_key(report_type), + "deduplicated": False, + "status": "operations_disabled", + } workflow, period_key, deduplicated = self.prepare(report_type, actor, force) if deduplicated: return { diff --git a/app/modules/workflows/constants.py b/app/modules/workflows/constants.py index de7539f..9449506 100644 --- a/app/modules/workflows/constants.py +++ b/app/modules/workflows/constants.py @@ -8,6 +8,7 @@ class WorkflowType(StrEnum): ENTERPRISE_ANALYTICS = "enterprise_analytics" AI_MEMORY_CAPTURE = "ai_memory_capture" LIFECYCLE_REPORT = "lifecycle_report" + MARKET_ANALYSIS = "market_analysis" class WorkflowStatus(StrEnum): diff --git a/app/tasks/__init__.py b/app/tasks/__init__.py index 275510d..83455f7 100644 --- a/app/tasks/__init__.py +++ b/app/tasks/__init__.py @@ -2,6 +2,7 @@ from app.tasks.app import celery_app from app.tasks import events as _events # noqa: F401 from app.tasks import legacy as _legacy # noqa: F401 from app.tasks import lifecycle as _lifecycle # noqa: F401 +from app.tasks import market as _market # noqa: F401 from app.tasks import reports as _reports # noqa: F401 from app.tasks import risk as _risk # noqa: F401 diff --git a/app/tasks/market.py b/app/tasks/market.py new file mode 100644 index 0000000..f50501b --- /dev/null +++ b/app/tasks/market.py @@ -0,0 +1,31 @@ +from datetime import date + +from app.core.background.task_queue.constants import TASK_RUN_MARKET_CLOSE +from app.core.database import SessionLocal +from app.modules.market.pipeline import MarketPipelineService +from app.tasks.app import celery_app + + +@celery_app.task( + name=TASK_RUN_MARKET_CLOSE, + autoretry_for=(Exception,), + retry_backoff=True, + retry_kwargs={"max_retries": 3}, +) +def run_market_report( + report_type: str = "close", trade_date: str | None = None, force: bool = False +) -> dict: + target = date.fromisoformat(trade_date) if trade_date else date.today() + db = SessionLocal() + try: + return MarketPipelineService(db).run( + report_type=report_type, + reference_date=target, + force=force, + actor="scheduler", + ) + finally: + db.close() + + +run_market_close = run_market_report diff --git a/app/tools/init_db.py b/app/tools/init_db.py index d6f5d43..ed50c55 100644 --- a/app/tools/init_db.py +++ b/app/tools/init_db.py @@ -7,6 +7,12 @@ from app.modules.business.models import ( Employee, FundAccount, LegacySyncRun, + MarketAnnouncement, + MarketDailyQuote, + MarketFinancialMetric, + MarketInstrument, + MarketMacroIndicator, + MarketWatchlist, PerformanceMetric, Policy, Procurement, @@ -51,6 +57,12 @@ _MODELS = [ RiskEvent, RiskEventAction, LegacySyncRun, + MarketAnnouncement, + MarketInstrument, + MarketDailyQuote, + MarketFinancialMetric, + MarketMacroIndicator, + MarketWatchlist, ReportPushRun, DomainEvent, WorkflowInstance, diff --git a/tests/test_smoke.py b/tests/test_smoke.py index 993ba21..a8935f4 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -30,6 +30,7 @@ os.environ["SCHEDULER_ENABLED"] = "false" from fastapi.testclient import TestClient from app.core.config import Settings, get_settings +from app.core.background.scheduler import create_scheduler from app.core.database import Base, SessionLocal, engine from app.core.http.pagination import bounded_limit, bounded_offset from app.core.security import require_api_key, require_audit_api_key @@ -56,6 +57,11 @@ from app.modules.business.models import ( ProjectContract, ProjectMember, ProjectMilestone, + MarketDailyQuote, + MarketAnnouncement, + MarketFinancialMetric, + MarketInstrument, + MarketMacroIndicator, ) from app.modules.business.service import _model_payload, serialize_model from app.modules.legacy_mysql.services import LegacyMySQLService @@ -101,6 +107,9 @@ from app.modules.feishu.commands import FeishuCommandService from app.modules.feishu.events import FeishuEventService from app.modules.feishu.constants import FeishuEventSource from app.modules.risk.constants import RiskEventActionValue +from app.modules.market.chart import render_market_chart +from app.modules.market.service import MarketService, TushareClient, normalize_symbol +from app.modules.market.pipeline import MarketPipelineService from app.modules.workflows.constants import WorkflowStatus, WorkflowType from app.modules.workflows.models import WorkflowInstance @@ -213,6 +222,8 @@ def test_feishu_webhook_routes_message_event() -> None: def test_feishu_rule_commands_create_list_disable_and_enable(monkeypatch) -> None: + monkeypatch.setenv("READ_ONLY_MODE", "false") + get_settings.cache_clear() monkeypatch.setattr( FeishuService, "send_text", @@ -256,9 +267,13 @@ def test_feishu_rule_commands_create_list_disable_and_enable(monkeypatch) -> Non assert rule.status == AIMemoryStatus.ACTIVE finally: db.close() + monkeypatch.delenv("READ_ONLY_MODE", raising=False) + get_settings.cache_clear() -def test_feishu_rule_command_preserves_sender_and_rejects_invalid_input() -> None: +def test_feishu_rule_command_preserves_sender_and_rejects_invalid_input(monkeypatch) -> None: + monkeypatch.setenv("READ_ONLY_MODE", "false") + get_settings.cache_clear() payload = { "schema": "2.0", "header": { @@ -272,9 +287,7 @@ def test_feishu_rule_command_preserves_sender_and_rejects_invalid_input() -> Non "chat_id": "oc_test", "message_id": "om_smoke_rule_actor_001", "message_type": "text", - "content": json.dumps( - {"text": "学习规则：风险建议先写事实依据再写行动"} - ), + "content": json.dumps({"text": "学习规则：风险建议先写事实依据再写行动"}), }, }, } @@ -287,9 +300,7 @@ def test_feishu_rule_command_preserves_sender_and_rejects_invalid_input() -> Non ) assert result["result"]["command"] == "rule_create" rule = db.execute( - select(AIMemoryEntry).where( - AIMemoryEntry.content == "风险建议先写事实依据再写行动" - ) + select(AIMemoryEntry).where(AIMemoryEntry.content == "风险建议先写事实依据再写行动") ).scalar_one() assert rule.actor == "ou_rule_teacher" @@ -310,6 +321,8 @@ def test_feishu_rule_command_preserves_sender_and_rejects_invalid_input() -> Non assert "已拒绝学习" in secret["content"] finally: db.close() + monkeypatch.delenv("READ_ONLY_MODE", raising=False) + get_settings.cache_clear() def test_v3_request_id_health_and_metrics() -> None: @@ -609,9 +622,10 @@ def test_dashboard_and_response_masking() -> None: assert masked_response.status_code == 200 masked_items = masked_response.json()["items"] assert any(item["code"] == "EXP-MASK-001" for item in masked_items) - assert next( - item["payment_account"] for item in masked_items if item["code"] == "EXP-MASK-001" - ) == "[MASKED]" + assert ( + next(item["payment_account"] for item in masked_items if item["code"] == "EXP-MASK-001") + == "[MASKED]" + ) dashboard_response = client.get("/api/v1/dashboard/summary", headers=headers) assert dashboard_response.status_code == 200 @@ -635,9 +649,7 @@ def test_configured_domain_response_masking(monkeypatch) -> None: list_response = client.get("/api/v1/business/expenses", headers=headers) assert list_response.status_code == 200 item = next( - item - for item in list_response.json()["items"] - if item["code"] == "EXP-MASK-CONFIG-001" + item for item in list_response.json()["items"] if item["code"] == "EXP-MASK-CONFIG-001" ) assert item["amount"] == "[MASKED]" finally: @@ -843,10 +855,7 @@ def test_project_lifecycle_report_summarizes_progress_cost_and_risk() -> None: assert data[LifecycleResponseKey.TITLE] == ReportTitle.PROJECT_LIFECYCLE assert data[LifecycleResponseKey.METRICS][LifecycleSection.PROJECTS][MetricKey.TOTAL] == 1 assert data[LifecycleResponseKey.METRICS][LifecycleSection.PROJECTS][MetricKey.DELAYED] == 1 - assert ( - data[LifecycleResponseKey.METRICS][LifecycleSection.PROJECTS][MetricKey.OVER_BUDGET] - == 1 - ) + assert data[LifecycleResponseKey.METRICS][LifecycleSection.PROJECTS][MetricKey.OVER_BUDGET] == 1 assert data[LifecycleResponseKey.METRICS][LifecycleSection.TASKS][MetricKey.OVERDUE] == 1 assert ( data[LifecycleResponseKey.METRICS][LifecycleSection.PROCUREMENTS][ @@ -855,9 +864,7 @@ def test_project_lifecycle_report_summarizes_progress_cost_and_risk() -> None: == 1 ) assert ( - data[LifecycleResponseKey.METRICS][LifecycleSection.EXPENSES][ - MetricKey.PENDING_APPROVAL - ] + data[LifecycleResponseKey.METRICS][LifecycleSection.EXPENSES][MetricKey.PENDING_APPROVAL] == 1 ) assert data[LifecycleResponseKey.METRICS][LifecycleSection.ATTENDANCE][MetricKey.ABNORMAL] == 1 @@ -875,10 +882,7 @@ def test_project_lifecycle_report_summarizes_progress_cost_and_risk() -> None: assert ai_response.status_code == 200 ai_data = ai_response.json() assert ai_data[LifecycleResponseKey.AI_ANALYSIS][AIResponseKey.OK] is True - assert ( - ai_data[LifecycleResponseKey.AI_ANALYSIS][AIResponseKey.PROVIDER] - == AIProviderName.NOOP - ) + assert ai_data[LifecycleResponseKey.AI_ANALYSIS][AIResponseKey.PROVIDER] == AIProviderName.NOOP def test_v3_enterprise_analytics_returns_read_only_sections() -> None: @@ -1078,10 +1082,7 @@ def test_report_push_failure_is_recorded() -> None: headers=headers, ) assert runs_response.status_code == 200 - assert any( - item["title"] == ReportTitle.DAILY_BRIEF - for item in runs_response.json()["items"] - ) + assert any(item["title"] == ReportTitle.DAILY_BRIEF for item in runs_response.json()["items"]) dashboard_response = client.get("/api/v1/dashboard/summary", headers=headers) assert dashboard_response.status_code == 200 @@ -1267,9 +1268,7 @@ def test_intasect_full_sync_marks_missing_projects_inactive() -> None: db = SessionLocal() try: IntasectSyncService(db, FakeSource([row])).sync_dataset("projects", "RUN-1") - project = db.execute( - select(Project).where(Project.external_id == "99101") - ).scalar_one() + project = db.execute(select(Project).where(Project.external_id == "99101")).scalar_one() assert project.is_active is True assert project.display_code == "B-99101" @@ -1353,12 +1352,12 @@ def test_personnel_lifecycle_does_not_treat_missing_ding_mapping_as_absence() -> def test_lifecycle_pipeline_is_idempotent(monkeypatch) -> None: + monkeypatch.setenv("READ_ONLY_MODE", "false") + get_settings.cache_clear() monkeypatch.setattr( IntasectSyncService, "sync_all", - lambda self, run_code, force_full=False, batch_size=500: { - "projects": {"processed": 1} - }, + lambda self, run_code, force_full=False, batch_size=500: {"projects": {"processed": 1}}, ) monkeypatch.setattr( ReportService, @@ -1383,6 +1382,8 @@ def test_lifecycle_pipeline_is_idempotent(monkeypatch) -> None: assert first["workflow_code"] == second["workflow_code"] finally: db.close() + monkeypatch.delenv("READ_ONLY_MODE", raising=False) + get_settings.cache_clear() def test_lifecycle_enqueue_respects_read_only_guard(monkeypatch) -> None: @@ -1418,12 +1419,12 @@ def test_lifecycle_report_api_validates_filters_and_report_type() -> None: def test_ai_unavailable_sends_notice_without_business_report(monkeypatch) -> None: + monkeypatch.setenv("READ_ONLY_MODE", "false") + get_settings.cache_clear() monkeypatch.setattr( IntasectSyncService, "sync_all", - lambda self, run_code, force_full=False, batch_size=500: { - "projects": {"processed": 1} - }, + lambda self, run_code, force_full=False, batch_size=500: {"projects": {"processed": 1}}, ) monkeypatch.setattr( ReportService, @@ -1455,6 +1456,8 @@ def test_ai_unavailable_sends_notice_without_business_report(monkeypatch) -> Non assert result["notified"] is True finally: db.close() + monkeypatch.delenv("READ_ONLY_MODE", raising=False) + get_settings.cache_clear() def test_user_rules_are_prioritized_in_ai_context(monkeypatch) -> None: @@ -1940,3 +1943,468 @@ def test_lifecycle_chart_renders_finance_section() -> None: } ) assert png.startswith(b"\x89PNG\r\n\x1a\n") + + +def test_market_sync_and_overview_use_deterministic_breadth() -> None: + class FakeMarketProvider: + def query(self, api_name, params, fields): + if api_name == "trade_cal": + return [{"exchange": "SSE", "cal_date": "20260710", "is_open": "1"}] + if api_name == "stock_basic": + return [ + { + "ts_code": "600101.SH", + "name": "Alpha", + "exchange": "SSE", + "industry": "科技", + "list_date": "20200101", + }, + { + "ts_code": "000101.SZ", + "name": "Beta", + "exchange": "SZSE", + "industry": "消费", + "list_date": "20200101", + }, + ] + if api_name == "daily": + return [ + { + "ts_code": "600101.SH", + "trade_date": "20260710", + "open": 10, + "high": 11, + "low": 9, + "close": 11, + "pre_close": 10, + "pct_chg": 10, + "vol": 100, + "amount": 200, + }, + { + "ts_code": "000101.SZ", + "trade_date": "20260710", + "open": 10, + "high": 10, + "low": 8, + "close": 9, + "pre_close": 10, + "pct_chg": -10, + "vol": 100, + "amount": 300, + }, + ] + if api_name == "daily_basic": + return [{"ts_code": "600101.SH", "pe_ttm": 20, "pb": 2, "total_mv": 10000}] + if api_name == "index_daily": + return [ + { + "ts_code": params["ts_code"], + "trade_date": "20260710", + "open": 100, + "high": 102, + "low": 99, + "close": 101, + "pre_close": 100, + "pct_chg": 1, + "vol": 10, + "amount": 20, + } + ] + return [] + + db = SessionLocal() + try: + result = MarketService(db, FakeMarketProvider()).sync_daily(date(2026, 7, 10)) + assert result == {"instruments": 2, "quotes": 6, "valuations": 1} + report = MarketService(db).market_overview(date(2026, 7, 10)) + assert report["data_available"] is True + assert report["metrics"]["advances"] == 1 + assert report["metrics"]["declines"] == 1 + assert report["metrics"]["limit_up"] == 1 + assert report["metrics"]["limit_down"] == 1 + assert report["metrics"]["turnover_cny"] == 500000 + finally: + db.close() + + +def test_stock_analysis_calculates_returns_risk_and_chart() -> None: + db = SessionLocal() + symbol = "600102.SH" + try: + db.add(MarketInstrument(symbol=symbol, name="Gamma", exchange="SSE", industry="金融")) + for index in range(70): + db.add( + MarketDailyQuote( + symbol=symbol, + trade_date=date(2026, 4, 1) + timedelta(days=index), + close_price=10 + index / 10, + pct_change=1, + pe=15, + pb=1.5, + ) + ) + db.add( + MarketFinancialMetric( + symbol=symbol, + period_end=date(2026, 3, 31), + revenue_yoy=12, + net_profit_yoy=8, + roe=10, + debt_to_assets=40, + operating_cashflow_yoy=None, + ) + ) + db.commit() + report = MarketService(db).stock_analysis("600102") + assert report["metrics"]["return_20d"] > 0 + assert report["metrics"]["max_drawdown"] == 0 + assert report["metrics"]["financial"]["revenue_yoy"] == 12 + assert report["metrics"]["financial"]["operating_cashflow_yoy"] is None + assert render_market_chart(report).startswith(b"\x89PNG\r\n\x1a\n") + assert normalize_symbol("000001") == "000001.SZ" + finally: + db.close() + + +def test_market_api_and_feishu_fail_closed_without_ai(monkeypatch) -> None: + monkeypatch.setenv("MARKET_ANALYSIS_ENABLED", "true") + get_settings.cache_clear() + try: + response = client.get( + "/api/v1/market/overview", + headers=headers, + params={"trade_date": "2026-07-10"}, + ) + assert response.status_code == 200 + assert response.json()["data_available"] is True + db = SessionLocal() + try: + result = FeishuCommandService(db).handle_text( + "市场分析", actor="ou_market", auto_reply=False + ) + assert result["command"] == "market_overview" + assert result["reply_type"] == "text" + assert "AI 当前不可用" in result["content"] + assert "领涨行业" not in result["content"] + + industry = FeishuCommandService(db).handle_text( + "行业分析 科技", actor="ou_market", auto_reply=False + ) + assert "Alpha" in industry["content"] + + comparison = FeishuCommandService(db).handle_text( + "股票对比 600101 000101", actor="ou_market", auto_reply=False + ) + assert "Alpha" in comparison["content"] + assert "Beta" in comparison["content"] + finally: + db.close() + finally: + monkeypatch.delenv("MARKET_ANALYSIS_ENABLED", raising=False) + get_settings.cache_clear() + + +def test_market_closed_day_skips_quote_collection() -> None: + class ClosedProvider: + def query(self, api_name, params, fields): + assert api_name == "trade_cal" + return [{"is_open": "0"}] + + db = SessionLocal() + try: + result = MarketService(db, ClosedProvider()).sync_daily(date(2026, 7, 11)) + assert result["market_closed"] == 1 + assert result["quotes"] == 0 + finally: + db.close() + + +def test_market_scheduler_registers_close_and_weekly_jobs(monkeypatch) -> None: + monkeypatch.setenv("MARKET_ANALYSIS_ENABLED", "true") + get_settings.cache_clear() + try: + scheduler = create_scheduler() + job_ids = {job.id for job in scheduler.get_jobs()} + assert "market_premarket_analysis" in job_ids + assert "market_close_analysis" in job_ids + assert "market_weekly_analysis" in job_ids + finally: + monkeypatch.delenv("MARKET_ANALYSIS_ENABLED", raising=False) + get_settings.cache_clear() + + +def test_market_ai_exception_is_reported_without_business_fallback(monkeypatch) -> None: + class BrokenAdapter: + provider_name = "broken" + + def ask(self, prompt, context): + raise TimeoutError("model timeout") + + monkeypatch.setenv("MARKET_ANALYSIS_ENABLED", "true") + get_settings.cache_clear() + monkeypatch.setattr("app.modules.ai_agent.service.get_adapter", lambda: BrokenAdapter()) + db = SessionLocal() + try: + result = FeishuCommandService(db).handle_text( + "市场分析", actor="ou_market", auto_reply=False + ) + assert result["reply_type"] == "text" + assert "AI 当前不可用" in result["content"] + assert "上涨" not in result["content"] + finally: + db.close() + monkeypatch.delenv("MARKET_ANALYSIS_ENABLED", raising=False) + get_settings.cache_clear() + + +def test_tushare_http_failure_has_stable_error(monkeypatch) -> None: + import httpx + + monkeypatch.setenv("MARKET_DATA_TOKEN", "test-token") + get_settings.cache_clear() + monkeypatch.setattr( + "app.modules.market.service.httpx.post", + lambda *args, **kwargs: (_ for _ in ()).throw(httpx.ConnectTimeout("timeout")), + ) + try: + with pytest.raises(HTTPException) as exc_info: + TushareClient().query("trade_cal", {}, ["is_open"]) + assert exc_info.value.status_code == 503 + assert exc_info.value.detail == "Market data provider unavailable" + finally: + monkeypatch.delenv("MARKET_DATA_TOKEN", raising=False) + get_settings.cache_clear() + + +def test_market_macro_and_announcement_sync_store_only_curated_fields() -> None: + class Provider: + def query(self, api_name, params, fields): + if api_name == "shibor": + return [{"date": "20260710", "on": "1.55"}] + if api_name == "cn_cpi": + return [{"month": "202606", "nt_yoy": "0.4"}] + if api_name == "cn_gdp": + return [{"quarter": "2026Q2", "gdp_yoy": "5.1"}] + if api_name == "cn_m": + return [{"month": "202606", "m2_yoy": "8.3"}] + if api_name == "anns_d" and params["ann_date"] == "20260710": + return [ + { + "ann_date": "20260710", + "ts_code": "600103.SH", + "title": "年度报告披露提示", + "url": "https://example.invalid/report.pdf", + "rec_time": "2026-07-10 08:30:00", + "content": "正文不得保存", + } + ] + return [] + + db = SessionLocal() + try: + service = MarketService(db, Provider()) + result = service.sync_macro(date(2026, 7, 10)) + assert result == {"shibor": 1, "cn_cpi": 1, "cn_gdp": 1, "cn_m": 1} + assert service.sync_announcements(date(2026, 7, 10), date(2026, 7, 10)) == 1 + codes = set(db.execute(select(MarketMacroIndicator.code)).scalars()) + assert {"SHIBOR_ON", "CPI_YOY", "GDP_YOY", "M2_YOY"}.issubset(codes) + announcement = db.execute( + select(MarketAnnouncement).where(MarketAnnouncement.symbol == "600103.SH") + ).scalar_one() + assert announcement.title == "年度报告披露提示" + assert not hasattr(announcement, "content") + assert service.sync_announcements(date(2026, 7, 10), date(2026, 7, 10)) == 1 + count = len( + list( + db.execute( + select(MarketAnnouncement).where( + MarketAnnouncement.symbol == "600103.SH" + ) + ).scalars() + ) + ) + assert count == 1 + finally: + db.close() + + +def test_market_pipeline_is_idempotent_and_requires_complete_ai_chart_delivery( + monkeypatch, +) -> None: + class FakeMarket: + def sync_daily(self, target): + return {"quotes": 2} + + def sync_macro(self, target): + return {"shibor": 1} + + def sync_announcements(self, start, end): + return 1 + + def sync_watchlist_financials(self): + return {} + + def market_overview(self, target=None, include_ai=False, actor="api"): + return { + "title": "市场收盘分析", + "data_available": True, + "metrics": {"advances": 1, "declines": 1, "flat": 0}, + "chart_data": {"advances": 1, "declines": 1, "flat": 0}, + "lines": ["- AI 已完成分析"], + "content": "AI 已完成分析", + "ai_analysis": {"ok": True, "answer": "分析完成"}, + } + + monkeypatch.setenv("FEISHU_APP_ID", "test-app") + monkeypatch.setenv("FEISHU_APP_SECRET", "test-secret") + monkeypatch.setenv("FEISHU_DEFAULT_CHAT_ID", "oc_market") + monkeypatch.setenv("READ_ONLY_MODE", "false") + get_settings.cache_clear() + monkeypatch.setattr( + "app.modules.feishu.service.FeishuService.upload_image", + lambda self, image, actor="system": {"data": {"image_key": "img-market"}}, + ) + monkeypatch.setattr( + "app.modules.feishu.service.FeishuService.send_card", + lambda self, card, receive_id=None, receive_id_type="chat_id", actor="system": { + "code": 0 + }, + ) + db = SessionLocal() + try: + pipeline = MarketPipelineService(db, FakeMarket()) + first = pipeline.run("close", date(2030, 7, 12), actor="pytest") + second = pipeline.run("close", date(2030, 7, 12), actor="pytest") + assert first["status"] == WorkflowStatus.COMPLETED + assert second["deduplicated"] is True + assert second["workflow_code"] == first["workflow_code"] + finally: + db.close() + monkeypatch.delenv("FEISHU_APP_ID", raising=False) + monkeypatch.delenv("FEISHU_APP_SECRET", raising=False) + monkeypatch.delenv("FEISHU_DEFAULT_CHAT_ID", raising=False) + monkeypatch.delenv("READ_ONLY_MODE", raising=False) + get_settings.cache_clear() + + +def test_market_pipeline_fails_closed_when_ai_is_unavailable(monkeypatch) -> None: + class FakeMarket: + def sync_daily(self, target): + return {"quotes": 1} + + def sync_macro(self, target): + return {} + + def sync_announcements(self, start, end): + return 0 + + def sync_watchlist_financials(self): + return {} + + def market_overview(self, target=None, include_ai=False, actor="api"): + return { + "title": "市场分析", + "data_available": True, + "metrics": {"advances": 1}, + "chart_data": {"advances": 1}, + "lines": ["- 不应推送"], + "content": "不应推送", + "ai_analysis": {"ok": False, "error": "timeout"}, + } + + monkeypatch.setenv("READ_ONLY_MODE", "false") + get_settings.cache_clear() + db = SessionLocal() + try: + result = MarketPipelineService(db, FakeMarket()).run( + "close", date(2030, 7, 15), actor="pytest" + ) + assert result["status"] == WorkflowStatus.FAILED + assert result["reason"] == "ai_unavailable" + workflow = db.execute( + select(WorkflowInstance).where(WorkflowInstance.code == result["workflow_code"]) + ).scalar_one() + assert workflow.current_step == "ai_unavailable" + finally: + db.close() + monkeypatch.delenv("READ_ONLY_MODE", raising=False) + get_settings.cache_clear() + + +def test_market_weekly_overview_uses_five_latest_trading_days() -> None: + db = SessionLocal() + try: + symbols = (("600104.SH", "Delta", "科技", 10, 12), ("000104.SZ", "Epsilon", "消费", 10, 8)) + days = [date(2031, 7, 7) + timedelta(days=index) for index in range(5)] + for symbol, name, industry, first, last in symbols: + db.add(MarketInstrument(symbol=symbol, name=name, industry=industry)) + for index, day in enumerate(days): + close = first + (last - first) * index / 4 + db.add(MarketDailyQuote(symbol=symbol, trade_date=day, close_price=close)) + db.commit() + report = MarketService(db).weekly_overview(date(2031, 7, 11)) + assert report["data_available"] is True + assert report["metrics"]["trading_days"] == 5 + assert report["metrics"]["advances"] == 1 + assert report["metrics"]["declines"] == 1 + assert report["metrics"]["period_start"] == "2031-07-07" + assert report["metrics"]["period_end"] == "2031-07-11" + finally: + db.close() + + +def test_feishu_market_rules_are_scoped_to_market(monkeypatch) -> None: + monkeypatch.setenv("MARKET_ANALYSIS_ENABLED", "true") + monkeypatch.setenv("READ_ONLY_MODE", "false") + get_settings.cache_clear() + db = SessionLocal() + try: + created = FeishuCommandService(db).handle_text( + "学习市场规则 90：市场结论必须列出数据日期", + actor="ou_market_rule", + auto_reply=False, + ) + assert "market / market" in created["content"] + listed = FeishuCommandService(db).handle_text( + "查看市场规则", actor="ou_market_rule", auto_reply=False + ) + assert "市场结论必须列出数据日期" in listed["content"] + assert "market/market" in listed["content"] + finally: + db.close() + monkeypatch.delenv("MARKET_ANALYSIS_ENABLED", raising=False) + monkeypatch.delenv("READ_ONLY_MODE", raising=False) + get_settings.cache_clear() + + +def test_read_only_mode_blocks_feishu_mutations_and_market_pipeline(monkeypatch) -> None: + class MustNotRunMarket: + def sync_daily(self, target): + pytest.fail("read-only mode must block market synchronization") + + monkeypatch.setenv("READ_ONLY_MODE", "true") + monkeypatch.setenv("MARKET_ANALYSIS_ENABLED", "true") + get_settings.cache_clear() + db = SessionLocal() + try: + rules = FeishuCommandService(db).handle_text( + "学习市场规则：这条规则不应写入", actor="ou_read_only", auto_reply=False + ) + watchlist = FeishuCommandService(db).handle_text( + "加入自选 600105", actor="ou_read_only", auto_reply=False + ) + pipeline = MarketPipelineService(db, MustNotRunMarket()).run( + "close", date(2032, 7, 12), actor="pytest" + ) + assert "只读模式" in rules["content"] + assert "只读模式" in watchlist["content"] + assert pipeline["status"] == "operations_disabled" + assert db.execute( + select(AIMemoryEntry).where(AIMemoryEntry.content == "这条规则不应写入") + ).scalar_one_or_none() is None + finally: + db.close() + monkeypatch.delenv("READ_ONLY_MODE", raising=False) + monkeypatch.delenv("MARKET_ANALYSIS_ENABLED", raising=False) + get_settings.cache_clear()