604 lines
22 KiB
Python
604 lines
22 KiB
Python
"""数据适配层 —— 桥接 qtdb_pro（Tushare 长表）与新量化模块（alpha158 / execution）。
|
||
|
||
背景（v1.2.0 数据对齐核查）：
|
||
- 现有 47 项目统一用**长表** `(stock_code, trade_date, close_price)`（qtdb_pro.hq_* 表）
|
||
- alpha_factors.py 需要**单股 Series**（close / open / high / low / volume / vwap）
|
||
- execution.py 需要**宽表**（date × stock_code）prices / volumes
|
||
- Tushare 字段命名：`ts_code / vol(手) / amount(千元) / pct_chg`，且**无 vwap 字段**
|
||
|
||
本模块提供可组合的数据适配函数，让新模块直接吃 qtdb_pro 真实数据：
|
||
1. `long_to_wide()` — 长表 → 宽表（date × stock_code）
|
||
2. `wide_to_long()` — 宽表 → 长表
|
||
3. `rename_tushare_columns()` — 列名映射（ts_code→stock_code, vol→volume 等）
|
||
4. `add_vwap_proxy()` — vwap 代理（Tushare 无 vwap 字段）
|
||
5. `apply_adj_factor()` — 复权（hq_daily × hq_adj_factor 前复权）
|
||
6. `prepare_stock_series()` — 单股提取（alpha_factors 输入）
|
||
7. `prepare_asset_return_snapshot()` — 带稳定 lineage 的资产日收益
|
||
8. `prepare_execution_inputs()` — execution 输入（prices + volumes 宽表）
|
||
9. `load_qtdb_daily()` — 便捷加载（qtdb_pro.hq_daily + 可选复权）
|
||
|
||
全部纯 pandas/numpy，零新依赖，mypy strict 兼容。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
from collections.abc import Mapping, Sequence
|
||
from dataclasses import dataclass
|
||
from datetime import date
|
||
from typing import Any
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
from quant_engine.logging import get_logger
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
__all__ = [
|
||
"AssetReturnSnapshot",
|
||
"long_to_wide",
|
||
"wide_to_long",
|
||
"rename_tushare_columns",
|
||
"add_vwap_proxy",
|
||
"apply_adj_factor",
|
||
"prepare_stock_series",
|
||
"prepare_asset_return_snapshot",
|
||
"prepare_execution_inputs",
|
||
"load_qtdb_daily",
|
||
]
|
||
|
||
# Tushare hq_daily 字段 → 统一命名
|
||
TUSHARE_RENAME: dict[str, str] = {
|
||
"ts_code": "stock_code",
|
||
"trade_date": "trade_date",
|
||
"open": "open",
|
||
"high": "high",
|
||
"low": "low",
|
||
"close": "close",
|
||
"pre_close": "pre_close",
|
||
"pct_chg": "pct_chg",
|
||
"vol": "volume", # 手
|
||
"amount": "amount", # 千元
|
||
}
|
||
|
||
|
||
@dataclass(frozen=True, slots=True, init=False, eq=False)
|
||
class AssetReturnSnapshot:
|
||
"""Immutable-by-interface daily return matrix with reproducible lineage."""
|
||
|
||
data_snapshot_id: str
|
||
source: str
|
||
source_snapshot_id: str
|
||
price_field: str
|
||
adjustment: str
|
||
return_method: str
|
||
start_date: date
|
||
end_date: date
|
||
sessions: int
|
||
assets: tuple[str, ...]
|
||
_returns: pd.DataFrame
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
data_snapshot_id: str,
|
||
source: str,
|
||
source_snapshot_id: str,
|
||
price_field: str,
|
||
adjustment: str,
|
||
return_method: str,
|
||
start_date: date,
|
||
end_date: date,
|
||
assets: tuple[str, ...],
|
||
returns: pd.DataFrame,
|
||
) -> None:
|
||
for value, name in (
|
||
(data_snapshot_id, "data_snapshot_id"),
|
||
(source, "source"),
|
||
(source_snapshot_id, "source_snapshot_id"),
|
||
(price_field, "price_field"),
|
||
(adjustment, "adjustment"),
|
||
(return_method, "return_method"),
|
||
):
|
||
if not isinstance(value, str) or not value.strip():
|
||
raise ValueError(f"{name} must be non-empty")
|
||
if returns.empty or not isinstance(returns.index, pd.DatetimeIndex):
|
||
raise ValueError("returns must contain a DatetimeIndex and at least one session")
|
||
if tuple(returns.columns) != assets:
|
||
raise ValueError("assets must match returns columns")
|
||
if start_date > end_date:
|
||
raise ValueError("start_date must not be after end_date")
|
||
|
||
object.__setattr__(self, "data_snapshot_id", data_snapshot_id.strip())
|
||
object.__setattr__(self, "source", source.strip())
|
||
object.__setattr__(self, "source_snapshot_id", source_snapshot_id.strip())
|
||
object.__setattr__(self, "price_field", price_field.strip())
|
||
object.__setattr__(self, "adjustment", adjustment.strip())
|
||
object.__setattr__(self, "return_method", return_method.strip())
|
||
object.__setattr__(self, "start_date", start_date)
|
||
object.__setattr__(self, "end_date", end_date)
|
||
object.__setattr__(self, "sessions", len(returns))
|
||
object.__setattr__(self, "assets", assets)
|
||
object.__setattr__(self, "_returns", returns.copy(deep=True))
|
||
|
||
@property
|
||
def returns(self) -> pd.DataFrame:
|
||
"""Return an isolated copy so callers cannot mutate the snapshot."""
|
||
return self._returns.copy(deep=True)
|
||
|
||
|
||
def _non_empty(value: str, name: str) -> str:
|
||
if not isinstance(value, str) or not value.strip():
|
||
raise ValueError(f"{name} must be non-empty")
|
||
return value.strip()
|
||
|
||
|
||
def _asset_return_snapshot_id(
|
||
prices: pd.DataFrame,
|
||
*,
|
||
source: str,
|
||
source_snapshot_id: str,
|
||
price_field: str,
|
||
adjustment: str,
|
||
) -> str:
|
||
values = prices.to_numpy(dtype=float, copy=True)
|
||
missing = np.isnan(values)
|
||
normalized = np.where(missing, 0.0, values).astype("<f8", copy=False)
|
||
metadata = {
|
||
"adjustment": adjustment,
|
||
"assets": [str(asset) for asset in prices.columns],
|
||
"price_field": price_field,
|
||
"return_method": "simple",
|
||
"schema": "asset-returns-v1",
|
||
"sessions": [timestamp.date().isoformat() for timestamp in prices.index],
|
||
"shape": list(values.shape),
|
||
"source": source,
|
||
"source_snapshot_id": source_snapshot_id,
|
||
}
|
||
digest = hashlib.sha256(
|
||
json.dumps(metadata, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
||
)
|
||
digest.update(missing.astype(np.uint8, copy=False).tobytes(order="C"))
|
||
digest.update(normalized.tobytes(order="C"))
|
||
return f"asset-returns-v1:{digest.hexdigest()}"
|
||
|
||
|
||
def long_to_wide(
|
||
df: pd.DataFrame,
|
||
value_col: str = "close",
|
||
date_col: str = "trade_date",
|
||
stock_col: str = "stock_code",
|
||
) -> pd.DataFrame:
|
||
"""长表 → 宽表（date × stock_code）。
|
||
|
||
Args:
|
||
df: 长表 DataFrame（含 date_col / stock_col / value_col 三列）
|
||
value_col: 值列名
|
||
date_col: 日期列名
|
||
stock_col: 股票代码列名
|
||
|
||
Returns:
|
||
宽表：index=trade_date（升序），columns=stock_code，值=value_col
|
||
|
||
Examples:
|
||
>>> df = pd.DataFrame({
|
||
... "stock_code": ["A", "A", "B", "B"],
|
||
... "trade_date": ["2024-01-01", "2024-01-02", "2024-01-01", "2024-01-02"],
|
||
... "close": [10.0, 11.0, 20.0, 21.0],
|
||
... })
|
||
>>> wide = long_to_wide(df)
|
||
>>> wide.shape
|
||
(2, 2)
|
||
"""
|
||
if df.empty:
|
||
return pd.DataFrame()
|
||
if date_col not in df.columns or stock_col not in df.columns or value_col not in df.columns:
|
||
raise ValueError(
|
||
f"long_to_wide: 缺少列 date_col={date_col} / stock_col={stock_col} / "
|
||
f"value_col={value_col}，实际列={list(df.columns)}"
|
||
)
|
||
wide = df.pivot(index=date_col, columns=stock_col, values=value_col)
|
||
wide.index = pd.to_datetime(wide.index)
|
||
return wide.sort_index()
|
||
|
||
|
||
def wide_to_long(
|
||
wide: pd.DataFrame,
|
||
value_name: str = "close",
|
||
date_col: str = "trade_date",
|
||
stock_col: str = "stock_code",
|
||
) -> pd.DataFrame:
|
||
"""宽表 → 长表（date × stock_code → 长表）。
|
||
|
||
Args:
|
||
wide: 宽表 DataFrame（index=date，columns=stock_code）
|
||
value_name: 值列名（如 'close' / 'volume'）
|
||
date_col: 日期列名（输出）
|
||
stock_col: 股票代码列名（输出）
|
||
|
||
Returns:
|
||
长表 DataFrame（date_col / stock_col / value_name 三列）
|
||
"""
|
||
if wide.empty:
|
||
return pd.DataFrame(columns=[date_col, stock_col, value_name])
|
||
long_df = wide.reset_index().melt(
|
||
id_vars=wide.index.name or "index",
|
||
var_name=stock_col,
|
||
value_name=value_name,
|
||
)
|
||
long_df = long_df.rename(columns={wide.index.name or "index": date_col})
|
||
return long_df.dropna(subset=[value_name])
|
||
|
||
|
||
def rename_tushare_columns(df: pd.DataFrame) -> pd.DataFrame:
|
||
"""Tushare 列名 → 统一命名（只重命名存在的列）。
|
||
|
||
Args:
|
||
df: qtdb_pro 查询结果 DataFrame
|
||
|
||
Returns:
|
||
rename 后的 DataFrame
|
||
|
||
Examples:
|
||
>>> df = pd.DataFrame({"ts_code": ["000001.SZ"], "vol": [100.0]})
|
||
>>> rename_tushare_columns(df).columns.tolist()
|
||
['stock_code', 'volume']
|
||
"""
|
||
rename_map = {k: v for k, v in TUSHARE_RENAME.items() if k in df.columns}
|
||
return df.rename(columns=rename_map)
|
||
|
||
|
||
def add_vwap_proxy(
|
||
df: pd.DataFrame,
|
||
method: str = "amount_vol",
|
||
) -> pd.DataFrame:
|
||
"""vwap 代理（Tushare hq_daily 无 vwap 字段）。
|
||
|
||
Args:
|
||
df: 含 open/high/low/close（必），volume/amount（method=amount_vol 时）
|
||
method:
|
||
- 'amount_vol': vwap = amount(千元) / volume(手) 换算为元/股
|
||
vwap = amount*1000 / (volume*100) = amount * 10 / volume
|
||
（Tushare amount 单位千元，vol 单位手=100股）
|
||
- 'typical': vwap = (high + low + close) / 3
|
||
- 'close': vwap = close
|
||
|
||
Returns:
|
||
新增 'vwap' 列的 DataFrame（原列保留）
|
||
|
||
Note:
|
||
amount_vol 更接近真实 VWAP（成交额/成交量）；
|
||
typical / close 为兜底（缺 amount 或 volume 时）。
|
||
"""
|
||
out = df.copy()
|
||
if method == "amount_vol" and {"amount", "volume"}.issubset(out.columns):
|
||
out["vwap"] = out["amount"] * 10.0 / out["volume"].replace(0, np.nan)
|
||
elif method == "typical" and {"high", "low", "close"}.issubset(out.columns):
|
||
out["vwap"] = (out["high"] + out["low"] + out["close"]) / 3.0
|
||
elif {"close"}.issubset(out.columns):
|
||
out["vwap"] = out["close"]
|
||
else:
|
||
raise ValueError(f"add_vwap_proxy: 列不足。method={method}，实际列={list(out.columns)}")
|
||
return out
|
||
|
||
|
||
def apply_adj_factor(
|
||
daily: pd.DataFrame,
|
||
adj_factor: pd.DataFrame,
|
||
adj_col: str = "adj_factor",
|
||
date_col: str = "trade_date",
|
||
stock_col: str = "stock_code",
|
||
mode: str = "qfq",
|
||
) -> pd.DataFrame:
|
||
"""复权：前复权（qfq）/ 后复权（hfq）。
|
||
|
||
Tushare 复权因子（hq_adj_factor.adj_factor）：
|
||
- 后复权价 = 原始价 × adj_factor
|
||
- 前复权价 = 原始价 × adj_factor / 最新 adj_factor（按股票）
|
||
|
||
Args:
|
||
daily: 原始行情（含 open/high/low/close/volume 等列 + date_col/stock_col）
|
||
adj_factor: 复权因子表（含 stock_col/date_col/adj_col 三列）
|
||
adj_col: 因子列名
|
||
date_col: 日期列名
|
||
stock_col: 股票代码列名
|
||
mode: 'qfq'（前复权，默认）/ 'hfq'（后复权）
|
||
|
||
Returns:
|
||
复权后的行情 DataFrame（新增 adj_factor 列；复权仅作用于 OHLC）
|
||
volume / amount 不复权（成交量本身不受价格复权影响）
|
||
|
||
Note:
|
||
- 只对 open/high/low/close/pre_close 复权
|
||
- 无 adj_factor 的日期保持原值（不 drop）
|
||
"""
|
||
if daily.empty:
|
||
return daily
|
||
if adj_factor.empty:
|
||
logger.warning("apply_adj_factor: adj_factor 为空，返回原始行情")
|
||
return daily
|
||
|
||
out = daily.copy()
|
||
adj = adj_factor[[stock_col, date_col, adj_col]].copy()
|
||
# 合并
|
||
merged = out.merge(adj, on=[stock_col, date_col], how="left")
|
||
# 每股"最新日期"因子（前复权基准：最后交易日 adj_factor）
|
||
# 先按日期排序，再取每股最后一条
|
||
merged = merged.sort_values([stock_col, date_col])
|
||
latest_map = merged.groupby(stock_col)[adj_col].last()
|
||
latest = merged[stock_col].map(latest_map)
|
||
if mode == "qfq":
|
||
ratio = merged[adj_col] / latest
|
||
elif mode == "hfq":
|
||
ratio = merged[adj_col]
|
||
else:
|
||
raise ValueError(f"apply_adj_factor: mode={mode} 不支持（qfq/hfq）")
|
||
ratio = ratio.fillna(1.0) # 无因子日期保持原值
|
||
|
||
for col in ("open", "high", "low", "close", "pre_close"):
|
||
if col in merged.columns:
|
||
merged[col] = merged[col] * ratio
|
||
return merged
|
||
|
||
|
||
def prepare_stock_series(
|
||
df: pd.DataFrame,
|
||
stock_code: str,
|
||
stock_col: str = "stock_code",
|
||
date_col: str = "trade_date",
|
||
) -> dict[str, pd.Series]:
|
||
"""单股提取 → alpha_factors 输入（close/open/high/low/volume/vwap Series）。
|
||
|
||
Args:
|
||
df: 长表行情（Tushare rename 后）
|
||
stock_code: 目标股票代码
|
||
stock_col: 股票代码列名
|
||
date_col: 日期列名
|
||
|
||
Returns:
|
||
dict[str, pd.Series]：{'close': Series, 'open': Series, ...}
|
||
index=trade_date（datetime），按日期升序
|
||
|
||
Note:
|
||
- 只返回存在的列（如无 volume 则不含 'volume' key）
|
||
- 自动补 vwap（若缺）：先尝试 amount/volume，否则 typical price
|
||
"""
|
||
if df.empty:
|
||
return {}
|
||
sub = df[df[stock_col] == stock_code].sort_values(date_col)
|
||
if sub.empty:
|
||
return {}
|
||
series_map: dict[str, pd.Series] = {}
|
||
for col in ("open", "high", "low", "close", "pre_close", "volume", "amount"):
|
||
if col in sub.columns:
|
||
s = sub.set_index(date_col)[col]
|
||
s.index = pd.to_datetime(s.index)
|
||
series_map[col] = s.astype(float)
|
||
# vwap 代理
|
||
if "vwap" not in series_map:
|
||
if {"amount", "volume"}.issubset(series_map):
|
||
series_map["vwap"] = (
|
||
series_map["amount"] * 10.0 / series_map["volume"].replace(0, np.nan)
|
||
)
|
||
elif {"high", "low", "close"}.issubset(series_map):
|
||
series_map["vwap"] = (
|
||
series_map["high"] + series_map["low"] + series_map["close"]
|
||
) / 3.0
|
||
elif "close" in series_map:
|
||
series_map["vwap"] = series_map["close"]
|
||
return series_map
|
||
|
||
|
||
def prepare_asset_return_snapshot(
|
||
df: pd.DataFrame,
|
||
*,
|
||
source: str,
|
||
source_snapshot_id: str,
|
||
price_col: str = "close",
|
||
adjustment: str = "none",
|
||
stock_col: str = "stock_code",
|
||
date_col: str = "trade_date",
|
||
) -> AssetReturnSnapshot:
|
||
"""Build deterministic simple daily returns from a long market-price table.
|
||
|
||
``source_snapshot_id`` must identify the upstream ingestion snapshot. The
|
||
resulting ID additionally fingerprints canonical price values and their
|
||
missing mask, so changed contents cannot retain the same downstream identity.
|
||
Missing prices are never forward-filled.
|
||
"""
|
||
normalized_source = _non_empty(source, "source")
|
||
normalized_source_snapshot_id = _non_empty(
|
||
source_snapshot_id,
|
||
"source_snapshot_id",
|
||
)
|
||
normalized_price_col = _non_empty(price_col, "price_col")
|
||
normalized_adjustment = _non_empty(adjustment, "adjustment")
|
||
if not isinstance(df, pd.DataFrame):
|
||
raise TypeError("df must be a pandas DataFrame")
|
||
if df.empty:
|
||
raise ValueError("df must contain market prices")
|
||
required = {date_col, stock_col, normalized_price_col}
|
||
missing_columns = sorted(required.difference(df.columns))
|
||
if missing_columns:
|
||
raise ValueError(f"prepare_asset_return_snapshot: missing columns={missing_columns}")
|
||
|
||
market = df[[date_col, stock_col, normalized_price_col]].copy()
|
||
if any(not isinstance(asset, str) or not asset.strip() for asset in market[stock_col]):
|
||
raise ValueError("asset labels must be non-empty strings")
|
||
market[stock_col] = market[stock_col].str.strip()
|
||
try:
|
||
normalized_dates = pd.to_datetime(market[date_col], errors="raise")
|
||
except (TypeError, ValueError) as error:
|
||
raise ValueError("trade dates must be valid dates") from error
|
||
if normalized_dates.isna().any():
|
||
raise ValueError("trade dates must be valid dates")
|
||
market[date_col] = normalized_dates.dt.normalize()
|
||
if market.duplicated(subset=[date_col, stock_col]).any():
|
||
raise ValueError("duplicate asset/session prices are not allowed")
|
||
|
||
try:
|
||
market[normalized_price_col] = pd.to_numeric(
|
||
market[normalized_price_col],
|
||
errors="raise",
|
||
)
|
||
except (TypeError, ValueError) as error:
|
||
raise ValueError("prices must be numeric") from error
|
||
observed_prices = market[normalized_price_col].dropna().to_numpy(dtype=float)
|
||
if observed_prices.size == 0 or not np.isfinite(observed_prices).all():
|
||
raise ValueError("prices must contain positive finite observations")
|
||
if (observed_prices <= 0.0).any():
|
||
raise ValueError("prices must contain positive finite observations")
|
||
|
||
prices = market.pivot(
|
||
index=date_col,
|
||
columns=stock_col,
|
||
values=normalized_price_col,
|
||
).sort_index()
|
||
prices = prices.reindex(sorted(str(asset) for asset in prices.columns), axis="columns")
|
||
prices = prices.astype(float)
|
||
if len(prices) < 2:
|
||
raise ValueError("market prices must contain at least two sessions")
|
||
returns = prices.pct_change(fill_method=None)
|
||
assets = tuple(str(asset) for asset in prices.columns)
|
||
snapshot_id = _asset_return_snapshot_id(
|
||
prices,
|
||
source=normalized_source,
|
||
source_snapshot_id=normalized_source_snapshot_id,
|
||
price_field=normalized_price_col,
|
||
adjustment=normalized_adjustment,
|
||
)
|
||
return AssetReturnSnapshot(
|
||
data_snapshot_id=snapshot_id,
|
||
source=normalized_source,
|
||
source_snapshot_id=normalized_source_snapshot_id,
|
||
price_field=normalized_price_col,
|
||
adjustment=normalized_adjustment,
|
||
return_method="simple",
|
||
start_date=prices.index[0].date(),
|
||
end_date=prices.index[-1].date(),
|
||
assets=assets,
|
||
returns=returns,
|
||
)
|
||
|
||
|
||
def prepare_execution_inputs(
|
||
df: pd.DataFrame,
|
||
stock_col: str = "stock_code",
|
||
date_col: str = "trade_date",
|
||
*,
|
||
price_col: str = "close",
|
||
) -> tuple[pd.DataFrame, pd.DataFrame]:
|
||
"""长表行情 → execution 输入（prices + volumes 宽表）。
|
||
|
||
Args:
|
||
df: 长表行情（含 close / volume 列，Tushare rename 后）
|
||
stock_col: 股票代码列名
|
||
date_col: 日期列名
|
||
price_col: 执行价字段，默认 close；防前视研究可显式选择下一交易日 open
|
||
|
||
Returns:
|
||
(prices_wide, volumes_wide)：
|
||
- prices_wide: date × stock_code，值=price_col
|
||
- volumes_wide: date × stock_code，值=volume（若无 volume 列则全 1.0）
|
||
|
||
Examples:
|
||
>>> df = pd.DataFrame({
|
||
... "stock_code": ["A", "A", "B", "B"],
|
||
... "trade_date": ["2024-01-01", "2024-01-02"] * 2,
|
||
... "close": [10.0, 11.0, 20.0, 21.0],
|
||
... "volume": [100.0, 200.0, 300.0, 400.0],
|
||
... })
|
||
>>> prices, volumes = prepare_execution_inputs(df)
|
||
>>> prices.shape
|
||
(2, 2)
|
||
"""
|
||
if df.empty:
|
||
return pd.DataFrame(), pd.DataFrame()
|
||
if price_col not in df.columns:
|
||
raise ValueError(f"prepare_execution_inputs: 缺 {price_col} 列，实际列={list(df.columns)}")
|
||
prices = long_to_wide(df, value_col=price_col, date_col=date_col, stock_col=stock_col)
|
||
if "volume" in df.columns:
|
||
volumes = long_to_wide(df, value_col="volume", date_col=date_col, stock_col=stock_col)
|
||
else:
|
||
logger.warning("prepare_execution_inputs: 无 volume 列，volumes 全 1.0")
|
||
volumes = pd.DataFrame(1.0, index=prices.index, columns=prices.columns)
|
||
return prices, volumes
|
||
|
||
|
||
def load_qtdb_daily(
|
||
stock_codes: Sequence[str],
|
||
begin_date: str,
|
||
end_date: str | None = None,
|
||
with_adj: bool = False,
|
||
vwap_method: str = "amount_vol",
|
||
) -> pd.DataFrame:
|
||
"""便捷加载：qtdb_pro.hq_daily 长表行情（Tushare rename + vwap + 可选复权）。
|
||
|
||
Args:
|
||
stock_codes: 股票代码列表
|
||
begin_date: 起始日期（YYYY-MM-DD）
|
||
end_date: 结束日期（可选）
|
||
with_adj: 是否 join hq_adj_factor 前复权
|
||
vwap_method: vwap 代理方法（amount_vol / typical / close）
|
||
|
||
Returns:
|
||
长表 DataFrame：
|
||
stock_code / trade_date / open / high / low / close / pre_close /
|
||
pct_chg / volume / amount / vwap（+ adj_factor 若 with_adj）
|
||
|
||
Note:
|
||
- 需要真实 ClickHouse 连接（.env 配置 CLICKHOUSE_QTDB_*）
|
||
- 无连接时返回空 DataFrame + 告警
|
||
"""
|
||
try:
|
||
from src.data.clickhouse import qtdb_query
|
||
except ImportError:
|
||
logger.warning("load_qtdb_daily: src.data.clickhouse 不可用，返回空表")
|
||
return pd.DataFrame()
|
||
|
||
where = "ts_code IN %(codes)s AND trade_date >= %(begin)s"
|
||
params: dict[str, Any] = {"codes": tuple(stock_codes), "begin": begin_date}
|
||
if end_date:
|
||
where += " AND trade_date <= %(end)s"
|
||
params["end"] = end_date
|
||
|
||
sql = (
|
||
"SELECT ts_code, trade_date, open, high, low, close, pre_close, "
|
||
"pct_chg, vol, amount "
|
||
f"FROM qtdb_pro.hq_daily WHERE {where} ORDER BY stock_code, trade_date"
|
||
)
|
||
try:
|
||
result = qtdb_query(sql, params)
|
||
if hasattr(result, "to_df"):
|
||
df = result.to_df()
|
||
else:
|
||
df = pd.DataFrame(result)
|
||
except Exception as e:
|
||
logger.warning("load_qtdb_daily: 查询失败（返回空表）: %s", e)
|
||
return pd.DataFrame()
|
||
|
||
if df.empty:
|
||
return df
|
||
|
||
df = rename_tushare_columns(df)
|
||
df = add_vwap_proxy(df, method=vwap_method)
|
||
|
||
if with_adj:
|
||
try:
|
||
adj_sql = (
|
||
"SELECT ts_code, trade_date, adj_factor "
|
||
f"FROM qtdb_pro.hq_adj_factor WHERE {where} ORDER BY stock_code, trade_date"
|
||
)
|
||
adj_result = qtdb_query(adj_sql, params)
|
||
adj_df = (
|
||
adj_result.to_df() if hasattr(adj_result, "to_df") else pd.DataFrame(adj_result)
|
||
)
|
||
if not adj_df.empty:
|
||
adj_df = adj_df.rename(columns={"ts_code": "stock_code"})
|
||
df = apply_adj_factor(df, adj_df, mode="qfq")
|
||
except Exception as e:
|
||
logger.warning("load_qtdb_daily: 复权失败（返回未复权）: %s", e)
|
||
return df
|