ZLData 数据采集模块 — 详细设计文档

from xtquant import xtdata
 
# 1. 连接本地QMT服务
connect_result = xtdata.connect()
# 注意：需先启动 miniQMT 客户端
 
# 2. 获取股票列表
all_codes = xtdata.get_stock_list_in_sector('沪深A股')
 
# 3. 获取行业分类
sw1_sectors = [s for s in xtdata.get_sector_list()
               if str(s).upper().startswith('SW1') and '加权' not in str(s)]

data = xtdata.get_market_data_ex(
    stock_list=[stock_code],
    period='1d',
    dividend_type='front'  # 前复权
)

df = ak.stock_zh_a_hist(
    symbol='600519',
    period="daily",
    start_date='20240101',
    end_date='20251231',
    adjust="qfq"  # 前复权
)

TurnoverRate = (Volume × 100 / FloatingShares) × 100%

def _get_float_shares(stock_code: str) -> float:
    """从QMT获取流通股本"""
    detail = xtdata.get_instrument_detail(stock_code)
    return float(detail.get('TotalShares', 0)) if detail else 0
 
float_shares = _get_float_shares(stock_code)
for idx, row in df.iterrows():
    vol = int(row['volume'])
    vol_shares = vol * 100  # 手 → 股
    turnover = round(vol_shares / float_shares * 100, 4) \
        if float_shares > 0 and vol > 0 else None

SELECT stock_code, MAX(trade_date) AS max_date
FROM trade_stock_daily
GROUP BY stock_code

COL_MAP = {
    '日期': 'date',
    '开盘': 'open',
    '收盘': 'close',
    '最高': 'high',
    '最低': 'low',
    '成交量': 'volume',
    '成交额': 'amount',
    '振幅': 'amplitude',
    '涨跌幅': 'pct_chg',
    '涨跌额': 'change',
    '换手率': 'turnover',
}
df = df.rename(columns=COL_MAP)
df['date'] = pd.to_datetime(df['date'])

def get_field(record: dict, field_names: list, default=None):
    """
    从记录中获取字段值，支持多个候选字段名。
    按优先级依次尝试，返回第一个非空值。
    """
    for name in field_names:
        val = record.get(name)
        if val is not None:
            return val
    return default

# 净利润：优先含少数股东损益版
net_profit = get_field(inc, ['net_profit_incl_min_int_inc', 'net_profit_excl_min_int_inc'])
 
# 总资产
total_assets = get_field(bal, ['tot_assets'])
 
# ROE：优先取PershareIndex预计算值，没有则手动计算
roe = get_field(ps, ['du_return_on_equity', 'equity_roe', 'net_roe'])
if roe is None and net_profit and total_equity:
    roe = safe_divide(net_profit, total_equity, pct=True)

def safe_divide(a, b, pct=False):
    """安全除法，b为0时返回None。pct=True时结果乘以100。"""
    if a is None or b is None:
        return None
    a, b = float(a), float(b)
    if b == 0:
        return None
    result = a / b
    return round(result * 100, 4) if pct else round(result, 4)

def calc_netprofit_yoy(records: list) -> float | None:
    """用年报数据计算最新一期净利润同比增长率。"""
    annual = [r for r in records if str(r.get('end_date', '')).endswith('1231')]
    if len(annual) < 2:
        return None
    annual = sorted(annual, key=lambda x: x['end_date'])
    profits = [float(r['net_profit']) for r in annual if r.get('net_profit') is not None]
    if len(profits) < 2:
        return None
    s = pd.Series(profits)
    return round(s.pct_change().iloc[-1] * 100, 2)

INSERT INTO trade_macro_indicator
    (indicator_date, cpi_yoy, ppi_yoy, pmi, data_source)
VALUES
    ('2026-02-28', 2.1, -1.3, 50.8, 'akshare')
ON DUPLICATE KEY UPDATE
    cpi_yoy = COALESCE(VALUES(cpi_yoy), cpi_yoy),
    ppi_yoy = COALESCE(VALUES(ppi_yoy), ppi_yoy),
    pmi = COALESCE(VALUES(pmi), pmi);

df = ak.bond_china_yield(symbol="中国国债收益率")
# 提取10年期收益率列
cn_10y = df[['date', '中国国债收益率10年']]
cn_10y.columns = ['rate_date', 'cn_bond_10y']

from concurrent.futures import ThreadPoolExecutor, as_completed
import threading
 
NUM_WORKERS = 8  # 8个线程并行
_print_lock = threading.Lock()  # 线程锁，防止打印乱序
 
# 增量跳过：查询今日已采集股票
skip_sql = """
    SELECT DISTINCT stock_code FROM trade_stock_news
    WHERE DATE(created_at) = CURDATE()
"""

Deviation = (Actual_EPS - Consensus_EPS) / Consensus_EPS

Upside = (Target_Price - Current_Price) / Current_Price

SELECT DISTINCT stock_code FROM trade_report_consensus
WHERE created_at >= DATE_SUB(NOW(), INTERVAL 7 DAY)

# prompts.yaml
calendar:
  search_catalysts: |
    你是一位顶级的宏观策略分析师。请联网搜索并整理从{start_date} 到 {end_date}
    可能对 A 股产生"定调性"或"脉冲性"影响的事件。
    你需要重点关注：
    1. 决策层动态：如政治局会议（4/7/10/12月）、中央经济工作会议。
    2. 产业节点：如华为/特斯拉的发布会、英伟达财报日、全球AI峰会。
    3. 市场规则：如MSCI/富时罗素指数调整生效日。
    请以 JSON 数组格式返回，每个事件包含 date, title, country, category, importance。

def safe_float(val) -> float | None:
    """安全转换为浮点数"""
    if val is None or str(val).strip() in ('', '--', 'None', 'nan'):
        return None
    try:
        return float(val)
    except (ValueError, TypeError):
        return None
 
def safe_divide(a, b, pct=False) -> float | None:
    """安全除法，b为0时返回None"""
    if a is None or b is None:
        return None
    a, b = float(a), float(b)
    if b == 0:
        return None
    result = a / b
    return round(result * 100, 4) if pct else round(result, 4)

def normalize_timetag(ts_val):
    """将xtquant的m_timetag转换为日期字符串"""
    if ts_val is None:
        return None
    s = str(ts_val).strip()
    if len(s) == 8 and s.isdigit():
        return s
    try:
        v = float(s)
        if v == 0:
            return None
        if v > 1e12:
            v = v / 1000  # 毫秒 → 秒
        return datetime.fromtimestamp(v).strftime('%Y%m%d')
    except (OSError, ValueError, TypeError):
        return None

-- 幂等写入示例
INSERT INTO trade_stock_daily
    (stock_code, trade_date, close_price, volume)
VALUES
    ('600519.SH', '2026-02-24', 1468.02, 25000)
ON DUPLICATE KEY UPDATE
    close_price = VALUES(close_price),
    volume = VALUES(volume);

-- 宏观数据更新（COALESCE保护）
INSERT INTO trade_macro_indicator
    (indicator_date, cpi_yoy, ppi_yoy, pmi, m2_yoy, data_source)
VALUES
    ('2026-02-28', NULL, -1.3, 50.8, 8.5, 'akshare')
ON DUPLICATE KEY UPDATE
    ppi_yoy = COALESCE(VALUES(ppi_yoy), ppi_yoy),
    pmi = COALESCE(VALUES(pmi), pmi),
    m2_yoy = COALESCE(VALUES(m2_yoy), m2_yoy),
    data_source = VALUES(data_source);

# 错误做法：循环逐条写入（5000次数据库连接）
for row in rows:
    cursor.execute(INSERT_SQL, row)
 
# 正确做法：批量写入（1次数据库连接，快10+倍）
cursor.executemany(INSERT_SQL, rows)

from apscheduler.schedulers.background import BackgroundScheduler
 
scheduler = BackgroundScheduler()
 
# 收盘后行情更新
scheduler.add_job(
    update_daily_market_data,
    trigger='cron',
    hour=15, minute=35,
    id='market_data_update'
)
 
# 夜间宏观数据
scheduler.add_job(
    update_macro_indicators,
    trigger='cron',
    hour=20, minute=0,
    id='macro_update'
)
 
# 新闻采集
scheduler.add_job(
    collect_all_news,
    trigger='cron',
    hour=20, minute=20,
    id='news_collect'
)

# QMT 配置
QMT_HOST=127.0.0.1
QMT_PORT=58000
 
# 数据库配置
DB_HOST=localhost
DB_PORT=1433
DB_NAME=zldata
DB_USER=zldata_app
DB_PASSWORD=your_secure_password
 
# akshare 配置
AKSHARE_REQUEST_TIMEOUT=30
AKSHARE_REQUEST_DELAY=0.5
AKSHARE_MAX_RETRIES=3
 
# Redis 配置
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0
 
# LLM 配置（催化剂发现）
DASHSCOPE_API_KEY=sk-your-key
LLM_MODEL=qwen-max
 
# 采集配置
MARKET_DATA_WORKERS=8
FINANCIAL_BATCH_SIZE=50
NEWS_WORKERS=8
REPORT_WORKERS=4

from pydantic_settings import BaseSettings
 
class DataCollectionSettings(BaseSettings):
    """数据采集模块配置"""
    # QMT
    qmt_host: str = "127.0.0.1"
    qmt_port: int = 58000
 
    # 数据库
    db_host: str = "localhost"
    db_port: int = 1433
    db_name: str = "zldata"
    db_user: str = "zldata_app"
    db_password: str = ""
 
    # 采集参数
    market_data_workers: int = 8
    financial_batch_size: int = 50
    news_workers: int = 8
    report_workers: int = 4
    request_timeout: int = 30
    request_delay: float = 0.5
    max_retries: int = 3
 
    # LLM
    dashscope_api_key: str = ""
    llm_model: str = "qwen-max"
 
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"
 
settings = DataCollectionSettings()

import structlog
 
logger = structlog.get_logger()
 
# 采集成功日志
logger.info("market_data_downloaded",
    stock_code="600519.SH",
    date_range=["2026-02-24", "2026-03-24"],
    record_count=21,
    data_source="qmt",
    duration_ms=1250
)
 
# 采集失败日志
logger.error("data_source_unavailable",
    stock_code="000001.SZ",
    data_source="qmt",
    error="Connection refused",
    fallback="switched to akshare"
)

{
  "qmt": { "status": "online", "last_check": "2026-04-15T08:30:00Z" },
  "akshare": { "status": "online", "last_check": "2026-04-15T08:30:00Z" },
  "database": { "status": "online", "last_check": "2026-04-15T08:30:00Z" }
}

{
  "jobs": [
    {
      "id": "job_12345",
      "type": "market",
      "start_time": "2026-04-15T15:35:00Z",
      "end_time": "2026-04-15T15:40:00Z",
      "status": "success",
      "affected_rows": 5000,
      "duration_ms": 300000,
      "message": "All stocks updated successfully"
    }
  ]
}

{
  "sql": "SELECT stock_code, trade_date, close_price FROM trade_stock_daily WHERE stock_code = '600519.SH' ORDER BY trade_date DESC LIMIT 10;"
}

{
  "columns": ["stock_code", "trade_date", "close_price"],
  "rows": [
    ["600519.SH", "2026-04-14", 1680.50],
    ["600519.SH", "2026-04-13", 1675.20]
  ],
  "row_count": 2,
  "execution_time_ms": 45
}

{
  "error": "Forbidden query. Only SELECT statements on allowed tables are permitted."
}