fix: fetcher 数据库路径可移植化（db.py get_conn 统一，去除 Windows 硬编码）

deploy/e2e_server_test.py

@@ -0,0 +1,29 @@
+# -*- coding: utf-8 -*-
+"""服务器端端到端验证：宇宙 40 只 → K线同步 → 打分 → 推荐卡（在服务器 venv 内执行）"""
+import sys, warnings, time
+sys.path.insert(0, '/opt/a_stock_timeline')
+warnings.filterwarnings('ignore')
+from src.fetcher.kline_fetcher import KlineFetcher
+from src.quant.engine import QuantEngine
+
+DB = '/opt/a_stock_timeline/data/a_stock.db'
+eng = QuantEngine(emit=lambda e: print('[emit]', e['kind'], flush=True), db_path=DB)
+codes, names = eng.universe(40)
+print('宇宙股票:', len(codes), flush=True)
+kf = KlineFetcher(DB)
+t0 = time.time()
+ok = 0
+for i, c in enumerate(codes):
+    ok += 1 if kf.sync(c, 'day', days=150) else 0
+    if (i + 1) % 10 == 0:
+        print('  进度 {}/{} 成功{} 用时{:.0f}s'.format(i + 1, len(codes), ok, time.time() - t0), flush=True)
+print('K线同步完成: 成功 {} 只，用时 {:.0f}s'.format(ok, time.time() - t0), flush=True)
+recs = eng.build_recommendations(10)
+print('=== 推荐卡（Top 10）===', flush=True)
+for r in recs:
+    print('★{stars} {name}({code}) 现价{price} 购入{buy_low}~{buy_high} 预计{er}% | {reason}'.format(
+        stars=r['stars'], name=r['name'], code=r['code'], price=r['price'],
+        buy_low=r['buy_low'], buy_high=r['buy_high'],
+        er=r['expected_return_pct'] if r['expected_return_pct'] is not None else '—',
+        reason=r['reason']), flush=True)
+print('E2E DONE', flush=True)

src/analysis/mine_patterns.py

@@ -12,8 +12,9 @@ import pandas as pd
 
 warnings.filterwarnings('ignore')
 
-DB = r'C:' + chr(92) + 'Users' + chr(92) + 'lookt' + chr(92) + 'a_stock_timeline' + chr(92) + 'data' + chr(92) + 'a_stock.db'
+import os
-OUT = r'E:' + chr(92) + 'Data' + chr(92) + 'skills' + chr(92) + 'a-stock-timeline-patterns' + chr(92) + 'references'
+DB = os.environ.get('MINE_DB', 'data/a_stock.db')
+OUT = os.environ.get('MINE_OUT_DIR', '../skills-out/references')
 
 IDX_NAMES = {'sh000001': '上证指数', 'sz399001': '深证成指', 'sh000300': '沪深300',
              'sz399006': '创业板指', 'sh000688': '科创50'}

src/fetcher/calendar_fetcher.py

@@ -1,13 +1,15 @@
 """交易日历采集 - AkShare tool_trade_date_hist_sina"""
 import warnings, logging, sqlite3
 from datetime import datetime
+from pathlib import Path
 import pandas as pd
+from src.storage.db import get_conn
 import akshare as ak
 
 logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
 log = logging.getLogger(__name__)
 
-DB = r'C:\Users\lookt\a_stock_timeline\data\a_stock.db'
+DB = str(Path(__file__).resolve().parent.parent.parent / 'data' / 'a_stock.db')
 
 def fetch_calendar():
     df = ak.tool_trade_date_hist_sina()
@@ -17,7 +19,7 @@ def fetch_calendar():
     df['trade_date'] = pd.to_datetime(df['trade_date'])
     df['is_trade_day'] = df['trade_date'].dt.dayofweek.apply(lambda x: 0 if x >= 5 else 1)
     df['trade_date'] = df['trade_date'].dt.strftime('%Y-%m-%d')
-    conn = sqlite3.connect(DB)
+    conn = get_conn()
     df.to_sql('trade_calendar', conn, if_exists='replace', index=False)
     conn.close()
     log.info(f'写入 trade_calendar: {len(df)} 行')

src/fetcher/cn_news_fetcher.py

@@ -1,13 +1,15 @@
 """A股新闻采集 - AkShare stock_news_em"""
 import warnings, logging, sqlite3
 from datetime import datetime
+from pathlib import Path
 import pandas as pd
+from src.storage.db import get_conn
 import akshare as ak
 
 logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
 log = logging.getLogger(__name__)
 
-DB = r'C:\Users\lookt\a_stock_timeline\data\a_stock.db'
+DB = str(Path(__file__).resolve().parent.parent.parent / 'data' / 'a_stock.db')
 
 def fetch_news():
     df = ak.stock_news_em(symbol='A股')
@@ -32,7 +34,7 @@ def fetch_news():
     df['inserted_at'] = datetime.now().isoformat()
     if 'id' not in df.columns:
         df.insert(0, 'id', range(1, len(df) + 1))
-    conn = sqlite3.connect(DB)
+    conn = get_conn()
     df.to_sql('news_cn', conn, if_exists='replace', index=False)
     conn.close()
     log.info(f'写入 news_cn: {len(df)} 行')

src/fetcher/fund_flow_fetcher.py

@@ -4,6 +4,7 @@ import warnings
 import logging
 from urllib.parse import urlparse
 
+from pathlib import Path
 import pandas as pd
 import requests
 
@@ -13,7 +14,7 @@ warnings.filterwarnings('ignore')
 logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
 log = logging.getLogger(__name__)
 
-DB = r'C:\Users\lookt\a_stock_timeline\data\a_stock.db'
+DB = str(Path(__file__).resolve().parent.parent.parent / 'data' / 'a_stock.db')
 
 # 出站请求域名白名单（SSRF 防护：仅 http + 白名单主机，禁重定向跟随）
 _ALLOWED_HOSTS = {'data.10jqka.com.cn'}

src/fetcher/global_index_fetcher.py

@@ -1,13 +1,15 @@
 """外盘指数采集 - AkShare index_us_stock_sina"""
 import warnings, logging, sqlite3
 from datetime import datetime
+from pathlib import Path
 import pandas as pd
+from src.storage.db import get_conn
 import akshare as ak
 
 logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
 log = logging.getLogger(__name__)
 
-DB = r'C:\Users\lookt\a_stock_timeline\data\a_stock.db'
+DB = str(Path(__file__).resolve().parent.parent.parent / 'data' / 'a_stock.db')
 
 # 要采集的外盘指数
 INDICES = [
@@ -17,7 +19,7 @@ INDICES = [
 ]
 
 def fetch_global_index():
-    conn = sqlite3.connect(DB)
+    conn = get_conn()
     results = []
     for code, symbol, name in INDICES:
         try: