From 50939212be784e56a9271737692943c9d3bd8078 Mon Sep 17 00:00:00 2001 From: TIANHE Date: Tue, 30 Dec 2025 21:02:38 +0800 Subject: [PATCH] new Signed-off-by: TIANHE --- README.md | 38 +++- README_CN.md | 110 ++++++++- README_JA.md | 45 +++- README_KO.md | 45 +++- README_TW.md | 73 +++++- backend_api_python/README.md | 15 +- backend_api_python/app/__init__.py | 35 +++ backend_api_python/app/config/database.py | 4 +- backend_api_python/app/routes/analysis.py | 6 +- backend_api_python/app/routes/settings.py | 21 ++ .../app/services/agents/analyst_agents.py | 8 +- .../app/services/agents/base_agent.py | 41 +++- .../app/services/agents/coordinator.py | 30 ++- .../app/services/agents/embedding.py | 91 ++++++++ .../app/services/agents/memory.py | 215 +++++++++++++++--- .../app/services/agents/reflection.py | 41 ++-- .../app/services/agents/reflection_worker.py | 67 ++++++ .../app/services/agents/researcher_agents.py | 16 +- .../app/services/agents/tools.py | 31 ++- .../app/services/agents/trader_agent.py | 8 +- backend_api_python/app/services/analysis.py | 8 +- backend_api_python/app/utils/db.py | 33 +-- backend_api_python/env.example | 28 +++ 23 files changed, 910 insertions(+), 99 deletions(-) create mode 100644 backend_api_python/app/services/agents/embedding.py create mode 100644 backend_api_python/app/services/agents/reflection_worker.py diff --git a/README.md b/README.md index 8fde868..f9d7c32 100644 --- a/README.md +++ b/README.md @@ -117,6 +117,42 @@ Your tireless team of analysts: - **Crypto/Stock Agent**: Specializes in technical and capital flow analysis for specific markets. - **Report Generation**: Automatically produces structured Daily/Weekly research reports. +### 2.1 🧠 Memory-Augmented Agents (Local RAG + Reflection Loop) +QuantDinger’s agents don’t start from scratch every time. The backend includes a **local memory store** and an optional **reflection/verification loop**: + +- **What it is**: RAG-style experience retrieval injected into agent prompts (NOT model fine-tuning). +- **Where it lives**: Local SQLite files under `backend_api_python/data/memory/` (privacy-first). + +```mermaid +flowchart TD + A[POST /api/analysis/multi] --> B[AnalysisService] + B --> C[AgentCoordinator] + C --> D[Build context: price/kline/news/indicators] + + subgraph Agents[Agents] + E[Analysts + Researchers + Trader] + end + + C --> E + E -->|get_memories()| M[(SQLite: role_memory.db)] + M -->|Top-K memories| E + E -->|memory_prompt injected into system prompt| LLM[LLMService (OpenRouter)] + + C --> R[ReflectionService.record_analysis] + R --> RR[(SQLite: reflection_records.db)] + W[ReflectionWorker (optional)] --> RR + W -->|auto-verify + learn| M + A2[POST /api/analysis/reflect] -->|manual learn| M +``` + +**Retrieval ranking (simplified)**: + +\[ +score = w_{sim}\cdot sim + w_{recency}\cdot recency + w_{returns}\cdot returns\_score +\] + +Config lives in `.env` (see `backend_api_python/env.example`): `ENABLE_AGENT_MEMORY`, `AGENT_MEMORY_TOP_K`, `AGENT_MEMORY_ENABLE_VECTOR`, `AGENT_MEMORY_HALF_LIFE_DAYS`, and `ENABLE_REFLECTION_WORKER`. + ### 3. Robust Strategy Runtime - **Thread-Based Executor**: Independent thread pool management for strategy execution. - **Auto-Restore**: Automatically resumes running strategies after system restarts. @@ -471,7 +507,7 @@ Use `backend_api_python/env.example` as a template. Common settings include: - **Auth**: `SECRET_KEY`, `ADMIN_USER`, `ADMIN_PASSWORD` - **Server**: `PYTHON_API_HOST`, `PYTHON_API_PORT`, `PYTHON_API_DEBUG` -- **Database**: `SQLITE_DATABASE_FILE` (optional; default is `backend_api_python/quantdinger.db`) +- **Database**: `SQLITE_DATABASE_FILE` (optional; default is `backend_api_python/data/quantdinger.db`) - **AI / LLM**: `OPENROUTER_API_KEY`, `OPENROUTER_MODEL`, timeouts - **Web search**: `SEARCH_PROVIDER`, `SEARCH_GOOGLE_*`, `SEARCH_BING_API_KEY` - **Proxy (optional)**: `PROXY_PORT` or `PROXY_URL` diff --git a/README_CN.md b/README_CN.md index 8be62af..b1d40eb 100644 --- a/README_CN.md +++ b/README_CN.md @@ -117,6 +117,113 @@ - **加密/股票智能体**：专注于特定市场的技术和资金流向分析。 - **报告生成**：自动产出结构化的日报/周报。 +### 2.1 🧠 AI 记忆增强系统（Memory-Augmented Agents） +QuantDinger 的多智能体不是“每次从零开始”。它内置了一个**本地记忆库 + 反思闭环**，让每个智能体在生成提示词（prompt）时能检索过往经验，并在事后验证/复盘后把结果写回记忆库。 + +- **本质**：RAG 风格的“经验检索增强”，**不是**训练/微调模型权重（零外部向量库依赖）。 +- **隐私**：所有记忆与反思记录默认落盘在本地 SQLite：`backend_api_python/data/memory/`。 + +#### 逻辑图（从请求到记忆闭环） + +```mermaid +flowchart TD + A[POST /api/analysis/multi] --> B[AnalysisService] + B --> C[AgentCoordinator] + + C --> D[构建上下文: price/kline/news/indicators] + + subgraph Agents[多智能体并行/串行工作流] + E1[MarketAnalyst] + E2[FundamentalAnalyst] + E3[NewsAnalyst] + E4[SentimentAnalyst] + E5[RiskAnalyst] + F1[BullResearcher] + F2[BearResearcher] + G[TraderAgent] + end + + C -->|Phase 1 并行| E1 + C -->|Phase 1 并行| E2 + C -->|Phase 1 并行| E3 + C -->|Phase 1 并行| E4 + C -->|Phase 1 并行| E5 + C -->|Phase 2 并行| F1 + C -->|Phase 2 并行| F2 + C -->|Phase 3| G + + subgraph MemDB[本地 SQLite 记忆库（按角色拆分）] + M1[(data/memory/market_analyst_memory.db)] + M2[(data/memory/fundamental_analyst_memory.db)] + M3[(data/memory/news_analyst_memory.db)] + M4[(data/memory/sentiment_analyst_memory.db)] + M5[(data/memory/risk_analyst_memory.db)] + M6[(data/memory/bull_researcher_memory.db)] + M7[(data/memory/bear_researcher_memory.db)] + M8[(data/memory/trader_agent_memory.db)] + end + + E1 <-->|get_memories / add_memory| M1 + E2 <-->|get_memories / add_memory| M2 + E3 <-->|get_memories / add_memory| M3 + E4 <-->|get_memories / add_memory| M4 + E5 <-->|get_memories / add_memory| M5 + F1 <-->|get_memories / add_memory| M6 + F2 <-->|get_memories / add_memory| M7 + G <-->|get_memories / add_memory| M8 + + C --> R[ReflectionService.record_analysis] + R --> RR[(data/memory/reflection_records.db)] + W[ReflectionWorker（可选，定时）] --> RR + W -->|到期验证 + 写回经验| M8 + A2[POST /api/analysis/reflect（手动复盘）] -->|reflect_and_learn| M8 +``` + +#### 1) 记忆是如何“注入提示词”的？ +每个 agent 在 `analyze()` 时会： +- **构造 situation**：例如 `"{market}:{symbol} fundamental analysis"`、`"{market}:{symbol} trading decision"` 等 +- **携带结构化 metadata**：`market/symbol/timeframe` + `memory_features`（价格、涨跌幅、技术指标等） +- **检索 Top-K 历史经验**：转成一段可读的 `memory_prompt` +- **拼进 system_prompt**：模型在做本次分析前先“读历史经验” + +你可以在这些文件里看到同样的模式： +- `backend_api_python/app/services/agents/base_agent.py`：`get_memories()` + `format_memories_for_prompt()` +- `backend_api_python/app/services/agents/*_agents.py`、`trader_agent.py`：把 `memory_prompt` 拼进 system prompt + +#### 2) 记忆检索算法（为什么“像”RAG？） +每个角色的记忆表保存（简化）： +- **situation / recommendation / result / returns** +- **market / symbol / timeframe / features_json** +- **embedding（可选 BLOB）**：本地“哈希向量”嵌入（无外部依赖） + +检索时会从最近 `AGENT_MEMORY_CANDIDATE_LIMIT` 条候选中打分排序： +\[ +score = w_{sim}\cdot sim + w_{recency}\cdot recency + w_{returns}\cdot returns\_score +\] + +- **sim**：默认用 embedding cosine，相同维度的本地哈希向量；没有 embedding 时退化为 difflib 文本相似度 +- **recency**：半衰期衰减（`AGENT_MEMORY_HALF_LIFE_DAYS`） +- **returns_score**：对收益做 `tanh` 压缩，避免极值支配排序 +- **timeframe 惩罚**：如果查询 timeframe 与记忆记录 timeframe 不一致，会额外扣分 + +#### 3) “学习”从哪里来？（两条写入通道） +- **自动反思（可选）**： + - 分析结束后，系统会把 BUY/SELL/HOLD 记录到 `reflection_records.db` + - 开启 `ENABLE_REFLECTION_WORKER=true` 后，后台线程会按 `REFLECTION_WORKER_INTERVAL_SEC` 轮询到期记录，拉取最新价格做验证，并把验证结果写回 `trader_agent_memory.db` +- **手动复盘（推荐）**： + - 调用 `POST /api/analysis/reflect`，把你的真实交易结果（returns/result）写回记忆库，用于后续决策增强 + +#### 4) 关键环境变量（`.env`） +- **ENABLE_AGENT_MEMORY**：是否启用记忆增强（默认 true） +- **AGENT_MEMORY_TOP_K**：每次注入的经验条数（默认 5） +- **AGENT_MEMORY_CANDIDATE_LIMIT**：候选池大小（默认 500） +- **AGENT_MEMORY_ENABLE_VECTOR**：是否启用 embedding cosine（默认 true；否则退化为文本相似） +- **AGENT_MEMORY_EMBEDDING_DIM**：哈希 embedding 维度（默认 256） +- **AGENT_MEMORY_HALF_LIFE_DAYS**：时间衰减半衰期（默认 30） +- **AGENT_MEMORY_W_SIM / W_RECENCY / W_RETURNS**：三项权重（默认 0.75 / 0.20 / 0.05） +- **ENABLE_REFLECTION_WORKER**：是否启用自动验证闭环（默认 false） +- **REFLECTION_WORKER_INTERVAL_SEC**：自动验证周期（默认 86400 秒） + ### 3. 稳健的策略运行时 - **基于线程的执行器**：独立的线程池管理策略执行。 - **自动恢复**：系统重启后自动恢复运行中的策略。 @@ -277,9 +384,8 @@ docker-compose down -v ```yaml volumes: - - ./backend_api_python/quantdinger.db:/app/quantdinger.db # 数据库 - ./backend_api_python/logs:/app/logs # 日志 - - ./backend_api_python/data:/app/data # 数据目录 + - ./backend_api_python/data:/app/data # 数据目录（包含 quantdinger.db） - ./backend_api_python/.env:/app/.env # 配置文件 ``` diff --git a/README_JA.md b/README_JA.md index e6000bc..a5515d3 100644 --- a/README_JA.md +++ b/README_JA.md @@ -117,6 +117,48 @@ - **暗号資産/株式エージェント**: 特定の市場のテクニカル分析と資金フロー分析に特化しています。 - **レポート生成**: 構造化された日次/週次リサーチレポートを自動的に作成します。 +### 2.1 🧠 記憶拡張エージェント（Memory-Augmented Agents） +QuantDinger のエージェントは毎回「ゼロから」ではありません。バックエンドに **ローカル記憶DB + 反省（検証）ループ** を内蔵し、過去の経験を検索して system prompt に注入します（RAG 風）。 + +- **これは何か**：経験検索によるプロンプト拡張（※モデルの学習/微調整ではありません） +- **保存先**：ローカル SQLite（`backend_api_python/data/memory/`） + +#### フロー図（リクエスト → 記憶閉ループ） + +```mermaid +flowchart TD + A[POST /api/analysis/multi] --> B[AnalysisService] + B --> C[AgentCoordinator] + C --> D[コンテキスト構築: price/kline/news/indicators] + + subgraph Agents[Agents] + E[Analysts + Researchers + Trader] + end + + C --> E + E -->|get_memories / add_memory| M[(SQLite: data/memory/*_memory.db)] + C --> R[ReflectionService.record_analysis] + R --> RR[(SQLite: data/memory/reflection_records.db)] + W[ReflectionWorker（任意）] --> RR + W -->|期限到来を検証し学習| M + A2[POST /api/analysis/reflect] -->|手動で学習| M +``` + +#### 検索ランキング（簡略） +\[ +score = w_{sim}\cdot sim + w_{recency}\cdot recency + w_{returns}\cdot returns\_score +\] + +#### 学習の入口 +- **手動復習（推奨）**：`POST /api/analysis/reflect` で実際の結果（returns/result）を記憶へ +- **自動検証（任意）**：`ENABLE_REFLECTION_WORKER=true`、`REFLECTION_WORKER_INTERVAL_SEC` で定期検証→記憶へ反映 + +#### 主な環境変数（`.env`） +- `ENABLE_AGENT_MEMORY`, `AGENT_MEMORY_TOP_K`, `AGENT_MEMORY_CANDIDATE_LIMIT` +- `AGENT_MEMORY_ENABLE_VECTOR`, `AGENT_MEMORY_EMBEDDING_DIM` +- `AGENT_MEMORY_HALF_LIFE_DAYS`, `AGENT_MEMORY_W_SIM`, `AGENT_MEMORY_W_RECENCY`, `AGENT_MEMORY_W_RETURNS` +- `ENABLE_REFLECTION_WORKER`, `REFLECTION_WORKER_INTERVAL_SEC` + ### 3. 堅牢な戦略ランタイム - **スレッドベースのエグゼキューター**: 戦略実行のための独立したスレッドプール管理。 - **自動復元**: システム再起動後に実行中の戦略を自動的に再開します。 @@ -263,9 +305,8 @@ docker-compose down -v ```yaml volumes: - - ./backend_api_python/quantdinger.db:/app/quantdinger.db # データベース - ./backend_api_python/logs:/app/logs # ログ - - ./backend_api_python/data:/app/data # データディレクトリ + - ./backend_api_python/data:/app/data # データディレクトリ（quantdinger.db を含む） - ./backend_api_python/.env:/app/.env # 設定ファイル ``` diff --git a/README_KO.md b/README_KO.md index aa78d63..e6cf0b7 100644 --- a/README_KO.md +++ b/README_KO.md @@ -117,6 +117,48 @@ - **암호화폐/주식 에이전트**: 특정 시장에 대한 기술적 분석 및 자금 흐름 분석을 전문으로 합니다. - **보고서 생성**: 구조화된 일일/주간 리서치 보고서를 자동으로 생성합니다. +### 2.1 🧠 메모리 증강 에이전트(Memory-Augmented Agents) +QuantDinger의 에이전트는 매번 “처음부터” 시작하지 않습니다. 백엔드에는 **로컬 메모리 DB + 리플렉션(검증) 루프**가 내장되어 있어, 과거 경험을 검색해 system prompt에 주입하는 RAG 스타일의 강화가 동작합니다. + +- **무엇인가**: 경험 검색 기반 프롬프트 강화(모델 파인튜닝/학습이 아님) +- **저장 위치**: 로컬 SQLite (`backend_api_python/data/memory/`) + +#### 흐름도(요청 → 메모리 폐루프) + +```mermaid +flowchart TD + A[POST /api/analysis/multi] --> B[AnalysisService] + B --> C[AgentCoordinator] + C --> D[컨텍스트 구성: price/kline/news/indicators] + + subgraph Agents[Agents] + E[Analysts + Researchers + Trader] + end + + C --> E + E -->|get_memories / add_memory| M[(SQLite: data/memory/*_memory.db)] + C --> R[ReflectionService.record_analysis] + R --> RR[(SQLite: data/memory/reflection_records.db)] + W[ReflectionWorker(옵션)] --> RR + W -->|만기 검증 + 학습 반영| M + A2[POST /api/analysis/reflect] -->|수동 학습| M +``` + +#### 검색 랭킹(간략) +\[ +score = w_{sim}\cdot sim + w_{recency}\cdot recency + w_{returns}\cdot returns\_score +\] + +#### 학습 경로 +- **수동 회고(추천)**: `POST /api/analysis/reflect`로 실제 결과(returns/result)를 메모리에 기록 +- **자동 검증(옵션)**: `ENABLE_REFLECTION_WORKER=true`, `REFLECTION_WORKER_INTERVAL_SEC`로 주기적 검증→메모리 반영 + +#### 주요 환경 변수(`.env`) +- `ENABLE_AGENT_MEMORY`, `AGENT_MEMORY_TOP_K`, `AGENT_MEMORY_CANDIDATE_LIMIT` +- `AGENT_MEMORY_ENABLE_VECTOR`, `AGENT_MEMORY_EMBEDDING_DIM` +- `AGENT_MEMORY_HALF_LIFE_DAYS`, `AGENT_MEMORY_W_SIM`, `AGENT_MEMORY_W_RECENCY`, `AGENT_MEMORY_W_RETURNS` +- `ENABLE_REFLECTION_WORKER`, `REFLECTION_WORKER_INTERVAL_SEC` + ### 3. 강력한 전략 런타임 - **스레드 기반 실행기**: 전략 실행을 위한 독립적인 스레드 풀 관리. - **자동 복구**: 시스템 재시작 후 실행 중인 전략을 자동으로 재개합니다. @@ -276,9 +318,8 @@ docker-compose down -v ```yaml volumes: - - ./backend_api_python/quantdinger.db:/app/quantdinger.db # 데이터베이스 - ./backend_api_python/logs:/app/logs # 로그 - - ./backend_api_python/data:/app/data # 데이터 디렉토리 + - ./backend_api_python/data:/app/data # 데이터 디렉토리(quantdinger.db 포함) - ./backend_api_python/.env:/app/.env # 구성 파일 ``` diff --git a/README_TW.md b/README_TW.md index 32ee143..3ea6d34 100644 --- a/README_TW.md +++ b/README_TW.md @@ -117,6 +117,76 @@ - **加密/股票智能體**：專注於特定市場的技術和資金流向分析。 - **報告生成**：自動產出結構化的日報/週報。 +### 2.1 🧠 AI 記憶增強系統（Memory-Augmented Agents） +QuantDinger 的多智能體不是「每次從零開始」。後端內建**本地記憶庫 + 反思閉環**：在生成提示詞（prompt）時檢索過往經驗並注入到 system prompt，並可在事後驗證/復盤後把結果寫回記憶庫。 + +- **本質**：RAG 風格「經驗檢索增強」，**不是**訓練/微調模型權重（零外部向量庫依賴）。 +- **隱私**：所有記憶與反思記錄預設落盤在本地 SQLite：`backend_api_python/data/memory/`。 + +#### 邏輯圖（從請求到記憶閉環） + +```mermaid +flowchart TD + A[POST /api/analysis/multi] --> B[AnalysisService] + B --> C[AgentCoordinator] + + C --> D[建構上下文: price/kline/news/indicators] + + subgraph Agents[多智能體並行/串行工作流] + E1[MarketAnalyst] + E2[FundamentalAnalyst] + E3[NewsAnalyst] + E4[SentimentAnalyst] + E5[RiskAnalyst] + F1[BullResearcher] + F2[BearResearcher] + G[TraderAgent] + end + + C -->|Phase 1 並行| E1 + C -->|Phase 1 並行| E2 + C -->|Phase 1 並行| E3 + C -->|Phase 1 並行| E4 + C -->|Phase 1 並行| E5 + C -->|Phase 2 並行| F1 + C -->|Phase 2 並行| F2 + C -->|Phase 3| G + + subgraph MemDB[本地 SQLite 記憶庫（按角色拆分）] + M1[(data/memory/*_memory.db)] + end + + E1 <-->|get_memories / add_memory| M1 + E2 <-->|get_memories / add_memory| M1 + E3 <-->|get_memories / add_memory| M1 + E4 <-->|get_memories / add_memory| M1 + E5 <-->|get_memories / add_memory| M1 + F1 <-->|get_memories / add_memory| M1 + F2 <-->|get_memories / add_memory| M1 + G <-->|get_memories / add_memory| M1 + + C --> R[ReflectionService.record_analysis] + R --> RR[(data/memory/reflection_records.db)] + W[ReflectionWorker（可選，定時）] --> RR + W -->|到期驗證 + 寫回經驗| M1 + A2[POST /api/analysis/reflect（手動復盤）] -->|reflect_and_learn| M1 +``` + +#### 檢索排序（簡化） +\[ +score = w_{sim}\cdot sim + w_{recency}\cdot recency + w_{returns}\cdot returns\_score +\] + +#### 兩條「學習」通道 +- **手動復盤（推薦）**：`POST /api/analysis/reflect` 寫入真實交易結果（returns/result） +- **自動反思（可選）**：`ENABLE_REFLECTION_WORKER=true` 後，背景任務按 `REFLECTION_WORKER_INTERVAL_SEC` 驗證到期記錄並寫回記憶 + +#### 关键環境變量（`.env`） +- `ENABLE_AGENT_MEMORY`, `AGENT_MEMORY_TOP_K`, `AGENT_MEMORY_CANDIDATE_LIMIT` +- `AGENT_MEMORY_ENABLE_VECTOR`, `AGENT_MEMORY_EMBEDDING_DIM` +- `AGENT_MEMORY_HALF_LIFE_DAYS`, `AGENT_MEMORY_W_SIM`, `AGENT_MEMORY_W_RECENCY`, `AGENT_MEMORY_W_RETURNS` +- `ENABLE_REFLECTION_WORKER`, `REFLECTION_WORKER_INTERVAL_SEC` + ### 3. 穩健的策略運行時 - **基於線程的執行器**：獨立的線程池管理策略執行。 - **自動恢復**：系統重啟後自動恢復運行中的策略。 @@ -277,9 +347,8 @@ docker-compose down -v ```yaml volumes: - - ./backend_api_python/quantdinger.db:/app/quantdinger.db # 數據庫 - ./backend_api_python/logs:/app/logs # 日誌 - - ./backend_api_python/data:/app/data # 數據目錄 + - ./backend_api_python/data:/app/data # 數據目錄（包含 quantdinger.db） - ./backend_api_python/.env:/app/.env # 配置文件 ``` diff --git a/backend_api_python/README.md b/backend_api_python/README.md index af099f1..9ea3068 100644 --- a/backend_api_python/README.md +++ b/backend_api_python/README.md @@ -80,10 +80,23 @@ Default address: `http://localhost:5000` ## Database (SQLite) -- Default file: `backend_api_python/quantdinger.db` (override via `SQLITE_DATABASE_FILE`) +- Default file: `backend_api_python/data/quantdinger.db` (override via `SQLITE_DATABASE_FILE`) - Tables are created/updated automatically on startup (see `app/utils/db.py`) - `qd_addon_config` exists for backward compatibility, but **this backend reads secrets from `.env` / OS env**, not from the database (see `app/utils/config_loader.py`) +## AI memory augmentation (local-only) + +This backend includes a lightweight, privacy-first **memory-augmented multi-agent** system: + +- Memory DBs (per role): `backend_api_python/data/memory/*_memory.db` +- Reflection DB (optional auto-verify loop): `backend_api_python/data/memory/reflection_records.db` +- API hooks: + - `POST /api/analysis/multi` (main entry) + - `POST /api/analysis/reflect` (manual learn from post-trade outcomes) +- Controls: see `.env` / `env.example`: + - `ENABLE_AGENT_MEMORY`, `AGENT_MEMORY_*` + - `ENABLE_REFLECTION_WORKER`, `REFLECTION_WORKER_INTERVAL_SEC` + ## Frontend integration (Vue dev server) The Vue dev server proxies `/api/*` to this backend by default: diff --git a/backend_api_python/app/__init__.py b/backend_api_python/app/__init__.py index 722abbf..a48a8c9 100644 --- a/backend_api_python/app/__init__.py +++ b/backend_api_python/app/__init__.py @@ -13,6 +13,7 @@ logger = get_logger(__name__) # Global singletons (avoid duplicate strategy threads). _trading_executor = None _pending_order_worker = None +_reflection_worker = None def get_trading_executor(): @@ -33,6 +34,39 @@ def get_pending_order_worker(): return _pending_order_worker +def get_reflection_worker(): + """Get the reflection verification worker singleton.""" + global _reflection_worker + if _reflection_worker is None: + from app.services.agents.reflection_worker import ReflectionWorker + _reflection_worker = ReflectionWorker() + return _reflection_worker + + +def start_reflection_worker(): + """ + Start the reflection worker if enabled. + + To enable it, set ENABLE_REFLECTION_WORKER=true. + """ + import os + enabled = os.getenv("ENABLE_REFLECTION_WORKER", "false").lower() == "true" + if not enabled: + logger.info("Reflection worker is disabled. Set ENABLE_REFLECTION_WORKER=true to enable.") + return + + # Avoid running twice with Flask reloader + debug = os.getenv("PYTHON_API_DEBUG", "false").lower() == "true" + if debug: + if os.environ.get("WERKZEUG_RUN_MAIN") != "true": + return + + try: + get_reflection_worker().start() + except Exception as e: + logger.error(f"Failed to start reflection worker: {e}") + + def start_pending_order_worker(): """Start the pending order worker (disabled by default in paper mode). @@ -128,6 +162,7 @@ def create_app(config_name='default'): # Startup hooks. with app.app_context(): start_pending_order_worker() + start_reflection_worker() restore_running_strategies() return app diff --git a/backend_api_python/app/config/database.py b/backend_api_python/app/config/database.py index 540ff95..4d8bcf2 100644 --- a/backend_api_python/app/config/database.py +++ b/backend_api_python/app/config/database.py @@ -51,9 +51,9 @@ class MetaSQLiteConfig(type): @property def DATABASE_FILE(cls): - # 默认放在 backend_api_python 根目录 + # 默认放在 backend_api_python/data 目录（更干净，也与 docker-compose 挂载一致） base_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - default_path = os.path.join(base_dir, 'quantdinger.db') + default_path = os.path.join(base_dir, 'data', 'quantdinger.db') return os.getenv('SQLITE_DATABASE_FILE', default_path) diff --git a/backend_api_python/app/routes/analysis.py b/backend_api_python/app/routes/analysis.py index b97075a..8ec144b 100644 --- a/backend_api_python/app/routes/analysis.py +++ b/backend_api_python/app/routes/analysis.py @@ -83,6 +83,7 @@ def multi_analysis(): language = detect_request_language(request, body=data, default='en-US') model = data.get('model', None) use_multi_agent = data.get('use_multi_agent', None) # None -> use backend default + timeframe = data.get('timeframe', '1D') if not symbol or not market: return jsonify({ @@ -101,7 +102,7 @@ def multi_analysis(): # Create analysis service instance (local-only; no paid credits) service = AnalysisService(use_multi_agent=use_multi_agent) - result = service.analyze(market, symbol, language, model=model) + result = service.analyze(market, symbol, language, model=model, timeframe=timeframe) # Persist as "completed" history (no paid credits in local mode). task_id = _store_task(market, symbol, model or '', language, 'completed', result=result, error_message='') @@ -247,13 +248,14 @@ def stream_analysis(): symbol = data.get('symbol', '') language = detect_request_language(request, body=data, default='en-US') use_multi_agent = data.get('use_multi_agent', None) + timeframe = data.get('timeframe', '1D') def generate(): try: yield f"data: {json.dumps({'status': 'started', 'message': 'Analysis started'})}\n\n" service = AnalysisService(use_multi_agent=use_multi_agent) - result = service.analyze(market, symbol, language) + result = service.analyze(market, symbol, language, timeframe=timeframe) yield f"data: {json.dumps({'status': 'completed', 'data': result})}\n\n" except Exception as e: diff --git a/backend_api_python/app/routes/settings.py b/backend_api_python/app/routes/settings.py index 82975d2..76e7c39 100644 --- a/backend_api_python/app/routes/settings.py +++ b/backend_api_python/app/routes/settings.py @@ -95,6 +95,27 @@ CONFIG_SCHEMA = { {'key': 'ENABLE_AI_ANALYSIS', 'label': '启用AI分析', 'type': 'boolean', 'default': 'True'}, ] }, + 'agent_memory': { + 'title': '记忆/反思配置', + 'items': [ + {'key': 'ENABLE_AGENT_MEMORY', 'label': '启用Agent记忆', 'type': 'boolean', 'default': 'True'}, + {'key': 'AGENT_MEMORY_ENABLE_VECTOR', 'label': '启用向量检索(本地)', 'type': 'boolean', 'default': 'True'}, + {'key': 'AGENT_MEMORY_EMBEDDING_DIM', 'label': 'Embedding维度', 'type': 'number', 'default': '256'}, + {'key': 'AGENT_MEMORY_TOP_K', 'label': '召回数量TopK', 'type': 'number', 'default': '5'}, + {'key': 'AGENT_MEMORY_CANDIDATE_LIMIT', 'label': '候选窗口大小', 'type': 'number', 'default': '500'}, + {'key': 'AGENT_MEMORY_HALF_LIFE_DAYS', 'label': '时间衰减半衰期(天)', 'type': 'number', 'default': '30'}, + {'key': 'AGENT_MEMORY_W_SIM', 'label': '相似度权重', 'type': 'number', 'default': '0.75'}, + {'key': 'AGENT_MEMORY_W_RECENCY', 'label': '时间权重', 'type': 'number', 'default': '0.20'}, + {'key': 'AGENT_MEMORY_W_RETURNS', 'label': '收益权重', 'type': 'number', 'default': '0.05'}, + ] + }, + 'reflection_worker': { + 'title': '自动反思验证Worker', + 'items': [ + {'key': 'ENABLE_REFLECTION_WORKER', 'label': '启用自动验证', 'type': 'boolean', 'default': 'False'}, + {'key': 'REFLECTION_WORKER_INTERVAL_SEC', 'label': '验证周期间隔(秒)', 'type': 'number', 'default': '86400'}, + ] + }, 'ai': { 'title': 'AI/LLM配置', 'items': [ diff --git a/backend_api_python/app/services/agents/analyst_agents.py b/backend_api_python/app/services/agents/analyst_agents.py index d1c7ff8..46121f2 100644 --- a/backend_api_python/app/services/agents/analyst_agents.py +++ b/backend_api_python/app/services/agents/analyst_agents.py @@ -125,7 +125,13 @@ class FundamentalAnalyst(BaseAgent): # Memory situation = f"{market}:{symbol} fundamental analysis" - memories = self.get_memories(situation, n_matches=2) + memory_meta = { + "market": market, + "symbol": symbol, + "timeframe": context.get("timeframe"), + "features": context.get("memory_features") or {}, + } + memories = self.get_memories(situation, n_matches=None, metadata=memory_meta) memory_prompt = self.format_memories_for_prompt(memories) lang_instruction = self._get_language_instruction(language) diff --git a/backend_api_python/app/services/agents/base_agent.py b/backend_api_python/app/services/agents/base_agent.py index beb8ac8..1ea27e0 100644 --- a/backend_api_python/app/services/agents/base_agent.py +++ b/backend_api_python/app/services/agents/base_agent.py @@ -3,6 +3,7 @@ """ from abc import ABC, abstractmethod from typing import Dict, Any, Optional, List +import os from app.utils.logger import get_logger logger = get_logger(__name__) @@ -36,7 +37,7 @@ class BaseAgent(ABC): """ pass - def get_memories(self, situation: str, n_matches: int = 2) -> List[Dict[str, Any]]: + def get_memories(self, situation: str, n_matches: Optional[int] = None, metadata: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: """ 从记忆中检索相似情况 @@ -47,8 +48,17 @@ class BaseAgent(ABC): Returns: 匹配的历史记录列表 """ + if n_matches is None: + try: + n_matches = int(os.getenv("AGENT_MEMORY_TOP_K", "5") or 5) + except Exception: + n_matches = 5 if self.memory: - return self.memory.get_memories(situation, n_matches=n_matches) + # New memory API supports metadata; older implementations will ignore extra args. + try: + return self.memory.get_memories(situation, n_matches=n_matches, metadata=metadata) + except TypeError: + return self.memory.get_memories(situation, n_matches=n_matches) return [] def format_memories_for_prompt(self, memories: List[Dict[str, Any]]) -> str: @@ -62,10 +72,25 @@ class BaseAgent(ABC): 格式化的字符串 """ if not memories: - return "无历史经验可参考。" - - formatted = "历史经验参考：\n" + return "No prior experience available." + + lines = ["Prior experience (most relevant first):"] for i, mem in enumerate(memories, 1): - formatted += f"{i}. {mem.get('recommendation', 'N/A')}\n" - - return formatted + rec = mem.get("recommendation") or "N/A" + res = mem.get("result") or "" + ret = mem.get("returns") + created_at = mem.get("created_at") + # Keep created_at as-is (SQLite string), but include it for traceability. + meta_bits = [] + if created_at: + meta_bits.append(f"at {created_at}") + if ret is not None and ret != "": + meta_bits.append(f"returns={ret}%") + meta_s = f" ({', '.join(meta_bits)})" if meta_bits else "" + + if res: + lines.append(f"{i}. {rec}{meta_s}\n outcome: {res}") + else: + lines.append(f"{i}. {rec}{meta_s}") + + return "\n".join(lines) diff --git a/backend_api_python/app/services/agents/coordinator.py b/backend_api_python/app/services/agents/coordinator.py index d472121..5137767 100644 --- a/backend_api_python/app/services/agents/coordinator.py +++ b/backend_api_python/app/services/agents/coordinator.py @@ -97,7 +97,7 @@ class AgentCoordinator: self.neutral_analyst = NeutralAnalyst() self.safe_analyst = SafeAnalyst() - def run_analysis(self, market: str, symbol: str, language: str = 'zh-CN', model: str = None) -> Dict[str, Any]: + def run_analysis(self, market: str, symbol: str, language: str = 'zh-CN', model: str = None, timeframe: str = "1D") -> Dict[str, Any]: """ Run the full multi-agent analysis workflow. """ @@ -111,9 +111,13 @@ class AgentCoordinator: current_price = tools.get_current_price(market, symbol) company_data = tools.get_company_data(market, symbol, language=language) + # Normalize timeframe + tf = (timeframe or "1D").strip() + # 2) Kline + fundamentals - kline_data = tools.get_stock_data(market, symbol, days=30) + kline_data = tools.get_stock_data(market, symbol, days=30, timeframe=tf) fundamental_data = tools.get_fundamental_data(market, symbol) + indicators = tools.calculate_technical_indicators(kline_data or []) # 3) News (Finnhub + web search) company_name = company_data.get('name', symbol) if company_data else symbol @@ -127,6 +131,7 @@ class AgentCoordinator: "fundamental_data": fundamental_data, "company_data": company_data, "news_data": news_data, + "indicators": indicators, } context = { @@ -134,6 +139,14 @@ class AgentCoordinator: "symbol": symbol, "language": language, "model": model, + "timeframe": tf, + # Compact structured features for memory retrieval. + "memory_features": { + "timeframe": tf, + "price": (current_price or {}).get("price"), + "changePercent": (current_price or {}).get("changePercent"), + "indicators": indicators, + }, "base_data": base_data } @@ -492,7 +505,18 @@ class AgentCoordinator: # Update trader memory if 'trader' in self.memories: self.memories['trader'].add_memory( - situation, recommendation, result, returns + situation, + recommendation, + result, + returns, + metadata={ + "market": market, + "symbol": symbol, + "timeframe": None, + "features": { + "source": "manual_reflect", + } + }, ) logger.info(f"Reflection completed: {market}:{symbol}, decision={decision}, returns={returns}") diff --git a/backend_api_python/app/services/agents/embedding.py b/backend_api_python/app/services/agents/embedding.py new file mode 100644 index 0000000..eb2024a --- /dev/null +++ b/backend_api_python/app/services/agents/embedding.py @@ -0,0 +1,91 @@ +""" +Lightweight embedding utilities for local-only deployments. + +We intentionally avoid heavyweight ML deps (torch/sentence-transformers) and external services. +This module provides a deterministic "hashed embedding" (similar to feature hashing): +- Tokenize text +- Hash tokens into a fixed-size dense vector +- L2 normalize + +It is not as semantically strong as modern transformer embeddings, but it enables: +- Vector storage in SQLite +- Cosine similarity retrieval +- Recency/return weighted ranking +""" + +from __future__ import annotations + +import math +import os +import re +import struct +import hashlib +from typing import List, Optional + + +_TOKEN_RE = re.compile(r"[A-Za-z0-9_]+", re.UNICODE) + + +def _tokenize(text: str) -> List[str]: + t = (text or "").lower() + return _TOKEN_RE.findall(t) + + +class EmbeddingService: + """Deterministic local embedding service.""" + + def __init__(self, dim: Optional[int] = None): + self.dim = int(dim or os.getenv("AGENT_MEMORY_EMBEDDING_DIM", "256") or 256) + if self.dim <= 0: + self.dim = 256 + + def embed(self, text: str) -> List[float]: + """ + Return a dense, L2-normalized embedding vector. + """ + vec = [0.0] * self.dim + tokens = _tokenize(text) + if not tokens: + return vec + + # Feature hashing with signed counts + for tok in tokens: + h = hashlib.blake2b(tok.encode("utf-8"), digest_size=8).digest() + # Use first 8 bytes as unsigned int + v = int.from_bytes(h, "little", signed=False) + idx = v % self.dim + sign = -1.0 if ((v >> 63) & 1) else 1.0 + vec[idx] += sign + + # L2 normalize + norm = math.sqrt(sum(x * x for x in vec)) or 1.0 + return [x / norm for x in vec] + + def to_bytes(self, vec: List[float]) -> bytes: + """ + Pack float vector into little-endian float32 bytes for SQLite BLOB storage. + """ + if not vec: + return b"" + return struct.pack("<" + "f" * len(vec), *[float(x) for x in vec]) + + def from_bytes(self, blob: bytes) -> List[float]: + if not blob: + return [] + n = len(blob) // 4 + if n <= 0: + return [] + return list(struct.unpack("<" + "f" * n, blob[: n * 4])) + + +def cosine_sim(a: List[float], b: List[float]) -> float: + """ + Cosine similarity for L2-normalized vectors. + If vectors are not normalized, this becomes a scaled dot product. + """ + if not a or not b: + return 0.0 + n = min(len(a), len(b)) + return float(sum(a[i] * b[i] for i in range(n))) + + diff --git a/backend_api_python/app/services/agents/memory.py b/backend_api_python/app/services/agents/memory.py index f0eec94..66b3e56 100644 --- a/backend_api_python/app/services/agents/memory.py +++ b/backend_api_python/app/services/agents/memory.py @@ -1,15 +1,29 @@ """ -智能体记忆系统 -使用 SQLite + 简单的文本相似度匹配 +Agent memory system (local-only). + +This module stores agent experiences in SQLite and retrieves relevant past cases +to inject into prompts (RAG-style). It does NOT finetune model weights. + +Retrieval (configurable): +- Vector similarity via deterministic local embeddings (default) +- Fallback to difflib text similarity when embeddings are missing + +Ranking combines: +- similarity +- recency decay (half-life) +- optional returns weight """ + import sqlite3 import json import os +import math from typing import List, Dict, Any, Optional -from datetime import datetime +from datetime import datetime, timezone import difflib from app.utils.logger import get_logger +from .embedding import EmbeddingService, cosine_sim logger = get_logger(__name__) @@ -34,6 +48,13 @@ class AgentMemory: db_path = os.path.join(db_dir, f'{agent_name}_memory.db') self.db_path = db_path + self.embedder = EmbeddingService() + self.enable_vector = os.getenv("AGENT_MEMORY_ENABLE_VECTOR", "true").lower() == "true" + self.candidate_limit = int(os.getenv("AGENT_MEMORY_CANDIDATE_LIMIT", "500") or 500) + self.half_life_days = float(os.getenv("AGENT_MEMORY_HALF_LIFE_DAYS", "30") or 30) + self.w_sim = float(os.getenv("AGENT_MEMORY_W_SIM", "0.75") or 0.75) + self.w_recency = float(os.getenv("AGENT_MEMORY_W_RECENCY", "0.20") or 0.20) + self.w_returns = float(os.getenv("AGENT_MEMORY_W_RETURNS", "0.05") or 0.05) self._init_database() def _init_database(self): @@ -49,22 +70,91 @@ class AgentMemory: recommendation TEXT NOT NULL, result TEXT, returns REAL, + market TEXT, + symbol TEXT, + timeframe TEXT, + features_json TEXT, + embedding BLOB, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') - - # 创建索引 + + # Best-effort migration for older DBs + # NOTE: For existing tables, we must add missing columns BEFORE creating indexes + # that reference them (otherwise we'll hit: "no such column: market"). + cursor.execute("PRAGMA table_info(memories)") + existing_cols = {row[1] for row in cursor.fetchall() or []} + for col, ddl in { + "market": "TEXT", + "symbol": "TEXT", + "timeframe": "TEXT", + "features_json": "TEXT", + "embedding": "BLOB", + }.items(): + if col not in existing_cols: + cursor.execute(f"ALTER TABLE memories ADD COLUMN {col} {ddl}") + + # 创建索引（放在迁移之后，兼容旧库） cursor.execute(''' CREATE INDEX IF NOT EXISTS idx_created_at ON memories(created_at) ''') + cursor.execute(''' + CREATE INDEX IF NOT EXISTS idx_market_symbol ON memories(market, symbol) + ''') conn.commit() conn.close() except Exception as e: logger.error(f"初始化记忆数据库失败: {e}") - - def add_memory(self, situation: str, recommendation: str, result: Optional[str] = None, returns: Optional[float] = None): + + def _now_utc(self) -> datetime: + return datetime.now(timezone.utc) + + def _parse_ts(self, ts_val: Any) -> Optional[datetime]: + if ts_val is None: + return None + if isinstance(ts_val, datetime): + return ts_val + s = str(ts_val) + try: + return datetime.fromisoformat(s.replace("Z", "")) + except Exception: + return None + + def _recency_score(self, created_at: Any) -> float: + dt = self._parse_ts(created_at) + if not dt: + return 0.0 + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + age_days = max(0.0, (self._now_utc() - dt).total_seconds() / 86400.0) + hl = max(0.1, float(self.half_life_days or 30.0)) + return float(math.exp(-math.log(2.0) * (age_days / hl))) + + def _returns_score(self, returns: Any) -> float: + try: + r = float(returns) + except Exception: + return 0.0 + return float(math.tanh(r / 10.0)) + + def _build_embed_text(self, situation: str, recommendation: str, result: Optional[str], features_json: Optional[str]) -> str: + return "\n".join([ + f"situation: {situation or ''}", + f"recommendation: {recommendation or ''}", + f"result: {result or ''}", + f"features: {features_json or ''}", + ]) + + def add_memory( + self, + situation: str, + recommendation: str, + result: Optional[str] = None, + returns: Optional[float] = None, + metadata: Optional[Dict[str, Any]] = None, + ): """ 添加记忆 @@ -73,15 +163,32 @@ class AgentMemory: recommendation: 建议/决策 result: 结果描述（可选） returns: 收益（可选） + metadata: Optional structured metadata (market/symbol/timeframe/features...) """ try: + meta = metadata or {} + market = (meta.get("market") or "").strip() or None + symbol = (meta.get("symbol") or "").strip() or None + timeframe = (meta.get("timeframe") or "").strip() or None + features = meta.get("features") if isinstance(meta, dict) else None + try: + features_json = json.dumps(features, ensure_ascii=False) if features is not None else None + except Exception: + features_json = None + + embedding_blob = None + if self.enable_vector: + text = self._build_embed_text(situation, recommendation, result, features_json) + vec = self.embedder.embed(text) + embedding_blob = self.embedder.to_bytes(vec) + conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' - INSERT INTO memories (situation, recommendation, result, returns) - VALUES (?, ?, ?, ?) - ''', (situation, recommendation, result, returns)) + INSERT INTO memories (situation, recommendation, result, returns, market, symbol, timeframe, features_json, embedding) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ''', (situation, recommendation, result, returns, market, symbol, timeframe, features_json, embedding_blob)) conn.commit() conn.close() @@ -89,7 +196,7 @@ class AgentMemory: except Exception as e: logger.error(f"添加记忆失败: {e}") - def get_memories(self, current_situation: str, n_matches: int = 2) -> List[Dict[str, Any]]: + def get_memories(self, current_situation: str, n_matches: int = 5, metadata: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: """ 检索相似记忆 @@ -106,11 +213,11 @@ class AgentMemory: # 获取所有记忆 cursor.execute(''' - SELECT id, situation, recommendation, result, returns, created_at + SELECT id, situation, recommendation, result, returns, created_at, market, symbol, timeframe, features_json, embedding FROM memories ORDER BY created_at DESC - LIMIT 100 - ''') + LIMIT ? + ''', (int(self.candidate_limit),)) all_memories = cursor.fetchall() conn.close() @@ -118,33 +225,71 @@ class AgentMemory: if not all_memories: return [] - # 计算相似度 - scored_memories = [] - for mem in all_memories: - mem_id, situation, recommendation, result, returns, created_at = mem - - # 使用简单的文本相似度 - similarity = difflib.SequenceMatcher( - None, - current_situation.lower(), - situation.lower() - ).ratio() - - scored_memories.append({ + meta = metadata or {} + tf = (meta.get("timeframe") or "").strip() + features = meta.get("features") if isinstance(meta, dict) else None + try: + q_features_json = json.dumps(features, ensure_ascii=False) if features is not None else None + except Exception: + q_features_json = None + + query_vec = [] + if self.enable_vector: + query_text = self._build_embed_text(current_situation, "", "", q_features_json) + query_vec = self.embedder.embed(query_text) + + ranked = [] + for row in all_memories: + ( + mem_id, + situation, + recommendation, + result, + returns, + created_at, + market, + symbol, + timeframe, + features_json, + embedding_blob, + ) = row + + sim = 0.0 + if self.enable_vector and embedding_blob: + try: + mem_vec = self.embedder.from_bytes(embedding_blob) + sim = cosine_sim(query_vec, mem_vec) + except Exception: + sim = 0.0 + else: + sim = difflib.SequenceMatcher(None, (current_situation or "").lower(), (situation or "").lower()).ratio() + + rec = self._recency_score(created_at) + ret = self._returns_score(returns) + + score = (self.w_sim * sim) + (self.w_recency * rec) + (self.w_returns * ret) + + if tf and timeframe and str(timeframe).strip() != tf: + score -= 0.15 + + ranked.append({ 'id': mem_id, 'matched_situation': situation, 'recommendation': recommendation, 'result': result, 'returns': returns, - 'similarity_score': similarity, - 'created_at': created_at + 'created_at': created_at, + 'market': market, + 'symbol': symbol, + 'timeframe': timeframe, + 'features_json': features_json, + 'score': float(score), + 'sim': float(sim), + 'recency': float(rec), }) - - # 按相似度排序 - scored_memories.sort(key=lambda x: x['similarity_score'], reverse=True) - - # 返回前 n_matches 个 - return scored_memories[:n_matches] + + ranked.sort(key=lambda x: x.get('score', 0.0), reverse=True) + return ranked[: max(0, int(n_matches or 0))] except Exception as e: logger.error(f"检索记忆失败: {e}") diff --git a/backend_api_python/app/services/agents/reflection.py b/backend_api_python/app/services/agents/reflection.py index f8b18f1..8d8de7c 100644 --- a/backend_api_python/app/services/agents/reflection.py +++ b/backend_api_python/app/services/agents/reflection.py @@ -91,7 +91,7 @@ class ReflectionService: conn.commit() conn.close() - logger.info(f"已记录分析用于反思: {market}:{symbol}, 将在 {check_days} 天后验证") + logger.info(f"Recorded analysis for reflection: {market}:{symbol}, will verify after {check_days} day(s)") except Exception as e: logger.error(f"记录分析失败: {e}") @@ -148,37 +148,52 @@ class ReflectionService: if decision == "BUY": if actual_return > 2.0: - result_desc = "准确：买入后价格上涨" + result_desc = "Correct: price rose after BUY" is_good_prediction = True elif actual_return < -2.0: - result_desc = "错误：买入后价格下跌" + result_desc = "Wrong: price fell after BUY" else: - result_desc = "中性：价格波动不大" + result_desc = "Neutral: limited price movement" elif decision == "SELL": if actual_return < -2.0: - result_desc = "准确：卖出后价格下跌" + result_desc = "Correct: price fell after SELL" is_good_prediction = True elif actual_return > 2.0: - result_desc = "错误：卖出后价格上涨" + result_desc = "Wrong: price rose after SELL" else: - result_desc = "中性：价格波动不大" + result_desc = "Neutral: limited price movement" else: # HOLD if -2.0 <= actual_return <= 2.0: - result_desc = "准确：持有期间波动不大" + result_desc = "Correct: limited movement during HOLD" is_good_prediction = True else: - result_desc = f"偏差：持有期间出现了较大波动 ({actual_return:.2f}%)" + result_desc = f"Deviated: large movement during HOLD ({actual_return:.2f}%)" # 4. 写入记忆系统 (Let the agent learn) - memory_situation = f"{market}:{symbol} 自动验证 (预测日期: {analysis_date})" - memory_recommendation = f"当时决策: {decision} (置信度 {confidence}), 理由: {reasoning[:50]}..." - memory_result = f"验证结果: {result_desc}, 实际收益: {actual_return:.2f}% (初始 {initial_price} -> 最新 {current_price})" + memory_situation = f"{market}:{symbol} auto-verified (analysis_date: {analysis_date})" + memory_recommendation = f"Decision: {decision} (confidence {confidence}), reasoning: {(reasoning or '')[:120]}" + memory_result = f"Verification: {result_desc}; return={actual_return:.2f}% (initial {initial_price} -> final {current_price})" trader_memory.add_memory( memory_situation, memory_recommendation, memory_result, - actual_return + actual_return, + metadata={ + "market": market, + "symbol": symbol, + "timeframe": "1D", + "features": { + "source": "auto_verify", + "decision": decision, + "confidence": confidence, + "initial_price": initial_price, + "final_price": current_price, + "analysis_date": str(analysis_date), + "result_desc": result_desc, + "is_good_prediction": bool(is_good_prediction), + }, + } ) # 5. 更新记录状态 diff --git a/backend_api_python/app/services/agents/reflection_worker.py b/backend_api_python/app/services/agents/reflection_worker.py new file mode 100644 index 0000000..3b8cb15 --- /dev/null +++ b/backend_api_python/app/services/agents/reflection_worker.py @@ -0,0 +1,67 @@ +""" +Background worker for automated reflection verification. + +This replaces the need for an external cron job in local deployments. +It periodically runs ReflectionService.run_verification_cycle(). + +Controls (env): +- ENABLE_REFLECTION_WORKER=true/false (default: false) +- REFLECTION_WORKER_INTERVAL_SEC (default: 86400) +""" + +from __future__ import annotations + +import os +import time +import threading +from typing import Optional + +from app.utils.logger import get_logger +from .reflection import ReflectionService + +logger = get_logger(__name__) + + +class ReflectionWorker: + def __init__(self): + self._thread: Optional[threading.Thread] = None + self._stop = threading.Event() + + def start(self): + if self._thread and self._thread.is_alive(): + return + + interval = int(os.getenv("REFLECTION_WORKER_INTERVAL_SEC", "86400") or 86400) + interval = max(60, interval) # at least 1 minute + + def _run(): + logger.info(f"Reflection worker started (interval={interval}s)") + svc = ReflectionService() + # Initial small delay to avoid fighting startup spikes + time.sleep(3) + while not self._stop.is_set(): + try: + svc.run_verification_cycle() + except Exception as e: + logger.error(f"Reflection worker cycle failed: {e}") + # Sleep in small steps to react to stop quickly + remaining = interval + while remaining > 0 and not self._stop.is_set(): + step = min(5, remaining) + time.sleep(step) + remaining -= step + logger.info("Reflection worker stopped") + + self._thread = threading.Thread(target=_run, name="ReflectionWorker", daemon=True) + self._thread.start() + + def stop(self): + self._stop.set() + t = self._thread + if t and t.is_alive(): + try: + t.join(timeout=5) + except Exception: + pass + + diff --git a/backend_api_python/app/services/agents/researcher_agents.py b/backend_api_python/app/services/agents/researcher_agents.py index cd2ed9e..c4dfe40 100644 --- a/backend_api_python/app/services/agents/researcher_agents.py +++ b/backend_api_python/app/services/agents/researcher_agents.py @@ -32,7 +32,13 @@ class BullResearcher(BaseAgent): # Memory situation = f"{market}:{symbol} bull case" - memories = self.get_memories(situation, n_matches=2) + memory_meta = { + "market": market, + "symbol": symbol, + "timeframe": context.get("timeframe"), + "features": context.get("memory_features") or {}, + } + memories = self.get_memories(situation, n_matches=None, metadata=memory_meta) memory_prompt = self.format_memories_for_prompt(memories) lang_instruction = self._get_language_instruction(language) @@ -121,7 +127,13 @@ class BearResearcher(BaseAgent): # Memory situation = f"{market}:{symbol} bear case" - memories = self.get_memories(situation, n_matches=2) + memory_meta = { + "market": market, + "symbol": symbol, + "timeframe": context.get("timeframe"), + "features": context.get("memory_features") or {}, + } + memories = self.get_memories(situation, n_matches=None, metadata=memory_meta) memory_prompt = self.format_memories_for_prompt(memories) lang_instruction = self._get_language_instruction(language) diff --git a/backend_api_python/app/services/agents/tools.py b/backend_api_python/app/services/agents/tools.py index b9a6ed4..448da78 100644 --- a/backend_api_python/app/services/agents/tools.py +++ b/backend_api_python/app/services/agents/tools.py @@ -82,27 +82,45 @@ class AgentTools: """Whether akshare is available at runtime.""" return bool(self._has_akshare and self._ak is not None) - def get_stock_data(self, market: str, symbol: str, days: int = 30) -> Optional[List[Dict[str, Any]]]: + def get_stock_data(self, market: str, symbol: str, days: int = 30, timeframe: str = "1d") -> Optional[List[Dict[str, Any]]]: """ Get daily Kline data for recent days (best-effort). Args: market: Market symbol: Symbol - days: Days + days: Days (for daily) / candle count hint (best-effort for intraday) + timeframe: Kline timeframe (best-effort). Common values: 1d, 1h, 4h, 1w Returns: List of OHLCV dicts or None """ try: klines = [] + tf = (timeframe or "1d").strip().lower() + # Normalize common UI values + tf_map = { + "1d": "1d", + "1day": "1d", + "d": "1d", + "1h": "1h", + "60m": "1h", + "4h": "4h", + "240m": "4h", + "1w": "1wk", + "1wk": "1wk", + "w": "1wk", + } + tf_yf = tf_map.get(tf, "1d") if market == 'USStock': end_date = datetime.now().strftime('%Y-%m-%d') start_date = (datetime.now() - timedelta(days=days + 5)).strftime('%Y-%m-%d') ticker = yf.Ticker(symbol) - df = ticker.history(start=start_date, end=end_date, interval="1d") + # yfinance supports limited intervals. Fallback to 1d when unsupported. + interval = tf_yf if tf_yf in ["1d", "1h", "1wk"] else "1d" + df = ticker.history(start=start_date, end=end_date, interval=interval) if not df.empty: df = df.tail(days).reset_index() @@ -121,7 +139,9 @@ class AgentTools: exchange = self._ccxt_exchange() symbol_pair = f'{symbol}/USDT' start_time = int((datetime.now() - timedelta(days=days)).timestamp()) - ohlcv = exchange.fetch_ohlcv(symbol_pair, '1d', since=start_time * 1000, limit=days) + # CCXT timeframes: 1d, 1h, 4h ... + ccxt_tf = tf if tf in ["1d", "1h", "4h"] else "1d" + ohlcv = exchange.fetch_ohlcv(symbol_pair, ccxt_tf, since=start_time * 1000, limit=days) if ohlcv: for candle in ohlcv: klines.append({ @@ -175,7 +195,8 @@ class AgentTools: start_date = (datetime.now() - timedelta(days=days + 5)).strftime('%Y-%m-%d') ticker = yf.Ticker(yf_symbol) - df = ticker.history(start=start_date, end=end_date, interval="1d") + interval = tf_yf if tf_yf in ["1d", "1h", "1wk"] else "1d" + df = ticker.history(start=start_date, end=end_date, interval=interval) if not df.empty: df = df.tail(days).reset_index() diff --git a/backend_api_python/app/services/agents/trader_agent.py b/backend_api_python/app/services/agents/trader_agent.py index 01e228f..01a5bd2 100644 --- a/backend_api_python/app/services/agents/trader_agent.py +++ b/backend_api_python/app/services/agents/trader_agent.py @@ -38,7 +38,13 @@ class TraderAgent(BaseAgent): # Memory situation = f"{market}:{symbol} trading decision" - memories = self.get_memories(situation, n_matches=2) + memory_meta = { + "market": market, + "symbol": symbol, + "timeframe": context.get("timeframe"), + "features": context.get("memory_features") or {}, + } + memories = self.get_memories(situation, n_matches=None, metadata=memory_meta) memory_prompt = self.format_memories_for_prompt(memories) lang_instruction = self._get_language_instruction(language) diff --git a/backend_api_python/app/services/analysis.py b/backend_api_python/app/services/analysis.py index 6290d6c..2a10ce5 100644 --- a/backend_api_python/app/services/analysis.py +++ b/backend_api_python/app/services/analysis.py @@ -37,8 +37,10 @@ class AnalysisService: # Lazy import to avoid circular imports from app.services.agents.coordinator import AgentCoordinator + import os + enable_memory = os.getenv("ENABLE_AGENT_MEMORY", "true").lower() == "true" self.coordinator = AgentCoordinator( - enable_memory=True, + enable_memory=enable_memory, max_debate_rounds=2 ) logger.info("Multi-agent coordinator initialized") @@ -50,7 +52,7 @@ class AnalysisService: finally: AnalysisService._initializing = False - def analyze(self, market: str, symbol: str, language: str = 'en-US', model: str = None) -> Dict[str, Any]: + def analyze(self, market: str, symbol: str, language: str = 'en-US', model: str = None, timeframe: str = "1D") -> Dict[str, Any]: """ Args: market: Market (AShare, USStock, HShare, Crypto, Forex, Futures) @@ -79,7 +81,7 @@ class AnalysisService: try: logger.info(f"Run coordinator: {market}:{symbol}") - agent_result = self.coordinator.run_analysis(market, symbol, language, model=model) + agent_result = self.coordinator.run_analysis(market, symbol, language, model=model, timeframe=timeframe) logger.info(f"Coordinator result keys: {list(agent_result.keys())}") diff --git a/backend_api_python/app/utils/db.py b/backend_api_python/app/utils/db.py index aaa9686..02af32e 100644 --- a/backend_api_python/app/utils/db.py +++ b/backend_api_python/app/utils/db.py @@ -12,8 +12,12 @@ from app.utils.logger import get_logger logger = get_logger(__name__) -# 数据库文件默认路径（兼容旧行为：backend_api_python/quantdinger.db） +# SQLite 主库路径解析 +# +# 目标默认行为：把主库放到 `backend_api_python/data/quantdinger.db` +# 兼容旧行为：老版本会在 `backend_api_python/quantdinger.db` 建库 _BASE_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) +_DEFAULT_DB_FILE = os.path.join(_BASE_DIR, 'data', 'quantdinger.db') _LEGACY_DB_FILE = os.path.join(_BASE_DIR, 'quantdinger.db') @@ -23,13 +27,13 @@ def _get_db_file() -> str: Priority: - SQLITE_DATABASE_FILE env (Docker 推荐：/app/data/quantdinger.db) - - legacy default: backend_api_python/quantdinger.db + - default: backend_api_python/data/quantdinger.db (local recommended) Also performs a best-effort one-time migration: - If the configured path doesn't exist but legacy db exists, copy legacy to configured path. + If the configured/default path doesn't exist but legacy db exists, copy legacy to configured/default path. """ env_path = os.getenv('SQLITE_DATABASE_FILE') - db_path = (env_path or '').strip() or _LEGACY_DB_FILE + db_path = (env_path or '').strip() or _DEFAULT_DB_FILE # Ensure parent dir exists parent = os.path.dirname(db_path) @@ -421,8 +425,9 @@ def _init_db_schema(conn): conn.commit() logger.info("Database schema initialized (SQLite)") -# 初始化一次 +# 初始化一次（按 db_file 维度） _has_initialized = False +_initialized_db_file = None class SQLiteCursor: """模拟 pymysql DictCursor""" @@ -492,24 +497,24 @@ def get_db_connection(): """ 获取数据库连接 (Context Manager) """ - global _has_initialized + global _has_initialized, _initialized_db_file # 简单的连接创建，不使用连接池（SQLite 文件锁机制决定了连接池意义不大） # 使用线程锁防止写冲突（虽然 SQLite 有 WAL 模式，但稳妥起见） # 注意：这里加锁粒度较大，如果是高并发场景可能会慢，但对于个人量化系统足够。 - # 初始化表结构 - if not _has_initialized: + # 初始化表结构（确保每个 db_file 都被初始化过） + db_file = _get_db_file() + if (not _has_initialized) or (_initialized_db_file != db_file): try: - db_file = _get_db_file() conn_init = sqlite3.connect(db_file) _init_db_schema(conn_init) conn_init.close() _has_initialized = True + _initialized_db_file = db_file except Exception as e: logger.error(f"Failed to initialize database: {e}") - db_file = _get_db_file() conn = SQLiteConnection(db_file) try: # with _db_lock: # SQLite 内部有锁，这里如果不跨线程共享连接其实不用强加锁 @@ -523,18 +528,18 @@ def get_db_connection(): def get_db_connection_sync(): """兼容旧接口""" - global _has_initialized - if not _has_initialized: + global _has_initialized, _initialized_db_file + db_file = _get_db_file() + if (not _has_initialized) or (_initialized_db_file != db_file): try: - db_file = _get_db_file() conn_init = sqlite3.connect(db_file) _init_db_schema(conn_init) conn_init.close() _has_initialized = True + _initialized_db_file = db_file except Exception as e: logger.error(f"Failed to initialize database: {e}") - db_file = _get_db_file() return SQLiteConnection(db_file) def close_db_connection(): diff --git a/backend_api_python/env.example b/backend_api_python/env.example index 1ab3d56..60bda7a 100644 --- a/backend_api_python/env.example +++ b/backend_api_python/env.example @@ -15,6 +15,14 @@ PYTHON_API_HOST=0.0.0.0 PYTHON_API_PORT=5000 PYTHON_API_DEBUG=False +# ========================= +# Database (SQLite) +# ========================= +# 主库文件路径（可选） +# - 不设置时，默认使用：backend_api_python/data/quantdinger.db +# - Docker 推荐：/app/data/quantdinger.db +SQLITE_DATABASE_FILE= + # ========================= # Pending orders worker (optional) # ========================= @@ -95,6 +103,26 @@ ENABLE_CACHE=False ENABLE_REQUEST_LOG=True ENABLE_AI_ANALYSIS=True +# ========================= +# Agent memory & reflection (optional) +# ========================= +# Toggle agent memory usage in multi-agent analysis +ENABLE_AGENT_MEMORY=true + +# Memory retrieval settings +AGENT_MEMORY_ENABLE_VECTOR=true +AGENT_MEMORY_EMBEDDING_DIM=256 +AGENT_MEMORY_TOP_K=5 +AGENT_MEMORY_CANDIDATE_LIMIT=500 +AGENT_MEMORY_HALF_LIFE_DAYS=30 +AGENT_MEMORY_W_SIM=0.75 +AGENT_MEMORY_W_RECENCY=0.20 +AGENT_MEMORY_W_RETURNS=0.05 + +# Automated verification loop (replaces cron) +ENABLE_REFLECTION_WORKER=false +REFLECTION_WORKER_INTERVAL_SEC=86400 + # ========================= # OpenRouter / LLM # =========================