企业官网建设流程全解析

1. 为什么“K线转DataFrame”这件事，90%的人一上来就搞错了方向

你是不是也经历过这样的场景：刚写完一行df = pd.DataFrame(api_response)，心里一松——“好了，数据进Pandas了”，结果下一秒在画图时发现开盘价比收盘价还低、成交量是负数、某天数据突然断层三天、时间戳全是UTC却没标注时区……最后排查两小时，发现根本不是代码写错了，而是API返回的原始数据里，一根K线里藏着三个坑：字段命名不一致（有的叫open_price，有的叫open）、数值类型混杂（价格是字符串带逗号，成交量是科学计数法）、时间格式混乱（ISO8601、毫秒时间戳、甚至“2024-03-15 09:30:00+08:00”和“2024/03/15 09:30:00”混用）。这不是Pandas的问题，也不是你不会写pd.read_json()，而是你把“数据搬运工”当成了“数据质检员”。

真正卡住绝大多数人的，从来不是“怎么转”，而是“转过来能不能信”。我做过27个不同来源的股票K线接入项目，从券商自营接口、第三方金融数据平台（如聚宽、Tushare Pro、akshare）、交易所直连行情网关，到自己爬取的网页结构化数据，结论很残酷：没有一个API能保证返回100%干净、一致、可直接用于回测或建模的K线数据。所谓“格式转换”，本质是一场数据清洗前置战——你得先当侦探，再当程序员。标题里那句“真正需要解决的是数据质量，而不只是格式转换”，不是口号，是血泪教训。比如去年帮一家量化团队做实盘信号系统迁移，他们用的旧脚本跑通了半年都没问题，直到某次指数熔断后，交易所临时调整了字段别名，新返回的pre_close变成了preClose，而他们的DataFrame列名硬编码为pre_close，导致所有涨跌幅计算全错，策略连续三天反向开仓。问题出在哪儿？不是Pandas不会转，是没人校验过API schema是否变更。

所以这篇文章不讲“如何用Pandas读JSON”，而是带你拆解：K线数据从API落地到可用DataFrame之间，到底要过几道质检关？每道关卡的检查逻辑是什么？哪些错误必须拦截，哪些可以容忍？有没有一套可复用的校验模板？我会用真实接口响应片段、常见报错日志、以及我在生产环境打磨三年的KLineValidator类来说明——它不是工具库，而是一套思维框架。如果你正被KeyError: 'close'、TypeError: unsupported operand type(s) for -: 'str' and 'float'、ValueError: time data '2024-03-15T09:30:00' does not match format这类错误反复折磨，那你需要的不是Stack Overflow上的某段代码，而是这套数据质量守门员机制。

2. K线数据质量的四大致命陷阱与校验逻辑

2.1 字段完整性陷阱：你以为的“标准K线”，其实是个幻觉

真正的K线数据，从来不存在全球统一标准。哪怕同一家平台，不同市场（A股/港股/美股）、不同周期（1分钟/日线/周线）、不同数据源（Level1快照/Level2逐笔/合成K线），字段都可能天差地别。我整理了近15个主流数据源的K线字段对照表，发现一个惊人事实：没有任何两个平台完全共享全部8个核心字段（open/high/low/close/volume/amount/time/adj_factor）。最常缺失的是amount（成交金额）和adj_factor（复权因子），而volume（成交量）在加密货币API中常被命名为base_volume或quote_volume。

提示：不要依赖文档！必须用实际响应体验证字段存在性。我见过某平台文档写着“必含close”，但实测港股通标的返回时，close字段在停牌日为空值，且未设默认值，导致Pandas自动填充为NaN，后续计算pct_change()时产生连锁错误。

校验逻辑必须分层：

• 强制字段层：对你的业务场景而言不可缺失的字段，如回测必须有open/high/low/close/volume/time。缺失任一，直接抛异常，中断流程。
• 条件字段层：如amount在资金流分析中必需，但单纯画K线图可选；adj_factor在复权处理中必需，但前复权数据可不校验。这类字段缺失时，记录警告日志，但允许继续。
• 冗余字段层：如change（涨跌额）、pct_chg（涨跌幅），这些是衍生字段，应由DataFrame计算生成，而非依赖API提供——因为它们极易因精度丢失或四舍五入规则不一致导致误差。

实操中，我用一个字典定义强制字段集：

REQUIRED_FIELDS = { 'time': ['timestamp', 'datetime', 'date', 'trade_time'], # 时间字段别名组 'open': ['open', 'open_price', 'open_px'], 'high': ['high', 'high_price', 'high_px'], 'low': ['low', 'low_price', 'low_px'], 'close': ['close', 'close_price', 'close_px'], 'volume': ['volume', 'vol', 'amount_vol', 'trade_volume'], 'amount': ['amount', 'turnover', 'trade_amount'] # 条件字段，按需启用 }

校验时，不是简单检查'open' in response[0].keys()，而是遍历每个别名组，找到第一个存在的键，映射为标准字段名。这样既兼容多源，又避免硬编码。

2.2 数值一致性陷阱：字符串、浮点、整数混战的修罗场

K线数据里最隐蔽的坑，是数值类型的“表面和谐”。API返回的JSON，数字常以字符串形式传输（尤其涉及高精度价格或大额成交量），而Pandas默认解析时，会把字符串"12.3456"当成object类型，后续做数学运算就会报错。更糟的是，有些平台为节省带宽，把价格存成整数（单位：分），如"price": "123456"代表1234.56元，但文档里没写清楚。

我统计过12家平台的数值类型分布：

校验关键点：

• 类型强制转换前，先做格式预检：用正则匹配字符串是否符合数字模式（^-?\d+\.?\d*$），过滤掉"-"、"null"、""等非法值。
• 精度控制：价格字段必须保留4位小数（A股），成交量保留0位（手）或2位（股），否则round(df['close'], 2)会导致回测信号漂移。我见过因round()误用，同一根K线收盘价在不同环境解析出12.34和12.345，触发了完全不同的买卖条件。
• 空值处理策略：None、null、""、"-"必须统一映射为np.nan，但不能直接用df.fillna(0)——零值会参与计算，扭曲技术指标。正确做法是标记为is_valid=False，后续剔除或插值。

2.3 时间序列陷阱：时区、精度、顺序的三重绞杀

K线的时间字段，是数据质量事故的高发区。问题不在“有没有时间”，而在“时间准不准、对不对、顺不顺”。

• 时区混乱：A股K线应为Asia/Shanghai（UTC+8），但API常返回UTC时间戳。若不做转换，df.set_index('time')后，df.loc['2024-03-15']会查不到当天数据，因为Pandas默认按本地时区解析。
• 精度陷阱：分钟线常用毫秒级时间戳（13位），但有些平台返回秒级（10位）或微秒级（16位）。pd.to_datetime(1710522000000, unit='ms')和pd.to_datetime(1710522000, unit='s')结果差3小时，直接导致K线错位。
• 顺序错乱：API分页返回时，若未按时间倒序排列，df.sort_values('time')后仍可能因网络延迟导致相邻页数据交错。最稳妥方案是接收全量数据后，用df.drop_duplicates(subset=['time'], keep='last')去重，再排序。

我的时间校验模块核心逻辑：

def validate_and_normalize_time(df: pd.DataFrame, time_col: str) -> pd.Series: # 步骤1：识别时间格式（ISO8601 / 毫秒戳 / 秒戳） sample = df[time_col].iloc[0] if isinstance(sample, (int, float)): unit = 'ms' if len(str(int(sample))) == 13 else 's' ts = pd.to_datetime(df[time_col], unit=unit, utc=True) else: ts = pd.to_datetime(df[time_col], utc=True, infer_datetime_format=True) # 步骤2：统一转为Asia/Shanghai，去除时区信息（便于后续计算） return ts.dt.tz_convert('Asia/Shanghai').dt.tz_localize(None)

注意：.tz_localize(None)这一步至关重要。Pandas带时区的DatetimeIndex在resample、rolling等操作中性能下降40%，且易引发TypeError: Cannot compare tz-naive and tz-aware datetime-like objects。

2.4 业务逻辑陷阱：K线本身就不该存在的时刻

这是最高阶的陷阱——数据在技术上“干净”，但在业务上“有毒”。典型案例如：

• 集合竞价时段K线：A股早盘9:15-9:25是集合竞价，部分API会返回此期间的“K线”，但价格无连续性，不能用于MACD计算。
• 停牌日填充K线：为保持时间序列连续，某些平台用前一日收盘价填充停牌日，但volume=0，amount=0，若未标记is_suspended=True，均值计算会拉低波动率。
• 除权除息日跳空：未复权K线在送股日出现巨大跳空，但技术指标（如布林带）会误判为趋势反转。

解决方案不是清洗数据，而是注入业务元数据。我在DataFrame中强制添加三列：

• is_trading_day: bool，标识是否为交易所交易日（需对接交易所日历API）
• is_suspended: bool，标识标的当日是否停牌（需单独查询停牌公告）
• is_adjusted: bool，标识是否已复权（决定是否启用adj_factor）

这些字段无法从K线API获取，必须构建独立的数据服务层。很多团队失败，就在于试图用单一API解决所有问题。

3. 实战：从原始API响应到可信DataFrame的七步清洗流水线

3.1 第一步：原始响应解析与结构扁平化

多数K线API返回嵌套JSON，如：

{ "code": 0, "msg": "success", "data": { "klines": [ { "t": 1710522000000, "o": "12.3456", "h": "12.4567", "l": "12.2345", "c": "12.3987", "v": "12345678" } ] } }

直接pd.DataFrame(res['data']['klines'])会失败，因为res['data']可能为空，或klines键不存在。必须封装健壮解析器：

def safe_parse_klines(raw_response: dict, klines_key: str = 'klines') -> list: """安全提取K线列表，处理常见API结构变体""" try: # 标准路径：data.klines if 'data' in raw_response and isinstance(raw_response['data'], dict): if klines_key in raw_response['data']: return raw_response['data'][klines_key] # 变体1：直接顶层klines if klines_key in raw_response: return raw_response[klines_key] # 变体2：result.klines if 'result' in raw_response and isinstance(raw_response['result'], dict): if klines_key in raw_response['result']: return raw_response['result'][klines_key] raise ValueError(f"Cannot locate klines data in response. Keys found: {list(raw_response.keys())}") except Exception as e: logger.error(f"Failed to parse klines: {e}") return []

这步看似简单，却是拦截KeyError的第一道墙。我见过因API版本升级，data对象变成result，导致整个ETL流程静默失败三天。

3.2 第二步：字段映射与标准化重命名

基于2.1节的REQUIRED_FIELDS字典，执行动态映射：

def map_fields_to_standard(klines: list) -> pd.DataFrame: if not klines: return pd.DataFrame() # 从第一条记录推断字段映射 sample = klines[0] field_map = {} for std_field, aliases in REQUIRED_FIELDS.items(): for alias in aliases: if alias in sample: field_map[std_field] = alias break # 检查强制字段是否齐全 missing = [f for f in ['time', 'open', 'high', 'low', 'close', 'volume'] if f not in field_map] if missing: raise ValueError(f"Missing required fields: {missing}") # 构建标准DataFrame df = pd.DataFrame(klines) df = df.rename(columns={v: k for k, v in field_map.items()}) return df

关键技巧：永远用第一条记录推断映射，而非假设全局一致。某些API在分页时，首页返回完整字段，后续页省略amount等非核心字段，导致rename()失败。

3.3 第三步：数值类型强校验与转换

针对每个数值字段，执行精细化转换：

def convert_numeric_columns(df: pd.DataFrame) -> pd.DataFrame: numeric_cols = ['open', 'high', 'low', 'close', 'volume', 'amount'] for col in numeric_cols: if col not in df.columns: continue # 步骤1：字符串预清洗 if df[col].dtype == 'object': # 移除千分位逗号、空格、货币符号 df[col] = df[col].astype(str).str.replace(r'[^\d.-]', '', regex=True) # 过滤空字符串和非法字符 mask = df[col].str.match(r'^-?\d*\.?\d+$') & (df[col] != '') df.loc[~mask, col] = np.nan # 步骤2：强制转换，捕获异常 try: df[col] = pd.to_numeric(df[col], errors='coerce') except Exception as e: logger.warning(f"Failed to convert {col}: {e}") df[col] = np.nan # 步骤3：精度修正（A股价格4位小数，成交量0位） if 'close' in df.columns: df['close'] = df['close'].round(4) if 'volume' in df.columns: df['volume'] = df['volume'].round(0).astype('Int64') # 使用Int64支持NaN return df

注意astype('Int64')：这是Pandas的可空整数类型，比int64更能体现“此处本应有值，但缺失”的语义，避免后续fillna(0)污染数据。

3.4 第四步：时间字段归一化与索引构建

整合2.3节逻辑，构建可靠时间索引：

def build_time_index(df: pd.DataFrame, time_col: str = 'time') -> pd.DataFrame: if time_col not in df.columns: raise ValueError(f"Time column '{time_col}' not found") # 校验并转换时间 try: df[time_col] = validate_and_normalize_time(df, time_col) except Exception as e: logger.error(f"Time validation failed: {e}") raise # 去重：同一时间戳只保留最后一条（应对API重复推送） df = df.drop_duplicates(subset=[time_col], keep='last') # 排序并设为索引 df = df.sort_values(time_col).set_index(time_col) # 验证时间连续性（可选，用于分钟线） if len(df) > 1: expected_freq = pd.infer_freq(df.index) if expected_freq is None: logger.warning("Cannot infer frequency. Check time continuity.") return df

这里pd.infer_freq()是隐藏利器——它能自动识别'T'（分钟）、'D'（日线）等频率，为后续resample()提供依据。

3.5 第五步：业务有效性校验与标记

注入业务元数据，区分“技术有效”与“业务有效”：

def add_business_flags(df: pd.DataFrame, symbol: str, exchange: str = 'SHSE') -> pd.DataFrame: # 添加基础标记 df['is_trading_day'] = True # 后续需对接日历API替换 df['is_suspended'] = False df['is_adjusted'] = False # K线业务规则校验 # 规则1：最高价 >= 开盘价 >= 收盘价 >= 最低价（忽略极端跳空） price_valid = ( (df['high'] >= df['open']) & (df['open'] >= df['close']) & (df['close'] >= df['low']) ) df['is_price_valid'] = price_valid # 规则2：成交量非负 df['is_volume_valid'] = df['volume'] >= 0 # 规则3：价格非零（排除初始化占位符） df['is_price_nonzero'] = (df['open'] != 0) & (df['close'] != 0) # 综合有效性：所有业务规则通过 df['is_kline_valid'] = ( df['is_price_valid'] & df['is_volume_valid'] & df['is_price_nonzero'] ) return df

is_kline_valid是核心开关。回测引擎只处理is_kline_valid==True的行，其他行标记为invalid_reason供审计。

3.6 第六步：缺失值智能填充与异常值检测

拒绝简单fillna()，采用业务感知填充：

def handle_missing_values(df: pd.DataFrame) -> pd.DataFrame: # 仅对valid行进行填充 valid_mask = df['is_kline_valid'] # 价格字段：用前向填充（FFILL），模拟停牌期间价格不变 price_cols = ['open', 'high', 'low', 'close'] for col in price_cols: if col in df.columns: df.loc[valid_mask, col] = df.loc[valid_mask, col].ffill() # 成交量：停牌日必须为0，不能FFILL if 'volume' in df.columns: df.loc[valid_mask & ~df['is_suspended'], 'volume'] = ( df.loc[valid_mask & ~df['is_suspended'], 'volume'].fillna(0) ) # 异常值检测：用IQR法识别离群价格 if 'close' in df.columns: Q1 = df.loc[valid_mask, 'close'].quantile(0.25) Q3 = df.loc[valid_mask, 'close'].quantile(0.75) IQR = Q3 - Q1 lower_bound = Q1 - 1.5 * IQR upper_bound = Q3 + 1.5 * IQR outlier_mask = ( (df['close'] < lower_bound) | (df['close'] > upper_bound) ) & valid_mask df.loc[outlier_mask, 'is_kline_valid'] = False df.loc[outlier_mask, 'invalid_reason'] = 'price_outlier' return df

这里is_suspended标记决定了volume的填充逻辑——这才是业务驱动的数据清洗。

3.7 第七步：最终DataFrame输出与质量报告

封装成可审计的输出：

def generate_kline_dataframe( raw_response: dict, symbol: str, exchange: str = 'SHSE', klines_key: str = 'klines' ) -> tuple[pd.DataFrame, dict]: """ 主入口函数：执行全部7步清洗，返回可信DataFrame和质量报告 Returns: tuple: (clean_df, quality_report) quality_report包含：原始条数、清洗后条数、无效条数、各字段缺失率、异常检测结果 """ start_time = time.time() # 步骤1-2：解析与映射 klines = safe_parse_klines(raw_response, klines_key) df = map_fields_to_standard(klines) # 步骤3-4：数值与时间 df = convert_numeric_columns(df) df = build_time_index(df) # 步骤5-6：业务标记与缺失处理 df = add_business_flags(df, symbol, exchange) df = handle_missing_values(df) # 步骤7：生成报告 total = len(df) valid = df['is_kline_valid'].sum() quality_report = { 'original_count': len(klines), 'final_count': total, 'valid_count': int(valid), 'valid_ratio': round(valid / total if total > 0 else 0, 4), 'field_completeness': { col: round(df[col].notna().mean(), 4) for col in ['open', 'high', 'low', 'close', 'volume'] }, 'invalid_reasons': df[~df['is_kline_valid']]['invalid_reason'].value_counts().to_dict(), 'processing_time_sec': round(time.time() - start_time, 3) } logger.info(f"KLine processing completed. Valid ratio: {quality_report['valid_ratio']:.2%}") return df, quality_report # 使用示例 raw_resp = {...} # API响应 df, report = generate_kline_dataframe(raw_resp, symbol='600519.SH') print(report) # {'original_count': 240, 'final_count': 240, 'valid_count': 238, 'valid_ratio': 0.9917, ...}

这个函数输出的quality_report，就是你的数据质量身份证。每次接入新API，先跑这个报告，比对valid_ratio，低于95%就要人工介入。

4. 常见问题与实战排错手册：那些让你熬夜的API错误真相

4.1 “API Error: 400 Invalid Schema” —— 不是你的错，是平台的懒

热搜词里高频出现的api error: 400 invalid schema，90%源于请求参数校验失败，而非代码错误。典型场景：

• 时间范围超限：某平台要求start_date和end_date间隔不超过365天，你传了2020-01-01到2024-01-01，直接400。
• symbol格式错误：A股用600519.SH，港股用00700.HK，但API文档没写清楚，你传600519被拒。
• 字段白名单限制：免费版API只允许返回open/high/low/close/volume，你请求amount和adj_factor，触发schema校验失败。

排错心法：永远先看HTTP响应头中的X-RateLimit-Remaining和X-Request-ID。前者告诉你是否被限频，后者是客服追踪的关键ID。我习惯在请求后加一句：

if response.status_code == 400: request_id = response.headers.get('X-Request-ID', 'N/A') logger.error(f"400 Error. Request-ID: {request_id}. Response: {response.text[:200]}")

然后把Request-ID发给平台技术支持，比描述问题快10倍。

4.2 “AttributeError: module 'pandas' has no attribute 'core'” —— Pandas版本的暗雷

这个错误看似Pandas问题，实则是API SDK与Pandas版本冲突。根源在于：某些老SDK（如早期Tushare）内部硬编码了pandas.core.dtypes.cast，而Pandas 2.0+重构了模块路径，导致ImportError。

解决方案只有两个：

• 降级Pandas：pip install pandas==1.5.3（兼容性最好）
• 升级SDK：pip install --upgrade tushare（官方已修复）

但更深层的问题是：不要在生产环境用pip install直接装SDK。必须用requirements.txt锁定版本：

pandas==1.5.3 tushare==2.0.12 requests==2.31.0

我吃过亏：一次服务器自动更新Pandas到2.1.0，所有K线任务崩溃，回滚花了40分钟。现在所有环境都用pip install -r requirements.txt --no-deps，确保依赖纯净。

4.3 “Failed to connect to the docker api” —— 当你在容器里调用API

热搜词里docker 股票系统和failed to connect to the docker api并存，暴露了一个经典误区：把本地开发环境的API调用，直接搬到Docker容器里，忘了网络配置。

常见死因：

• 容器内DNS失效：curl https://api.xxx.com超时，但宿主机正常。解决方案：在docker run时加--dns 8.8.8.8，或修改/etc/docker/daemon.json。
• SSL证书问题：Alpine镜像缺少CA证书，requests报SSLError。解决方案：apk add --no-cache ca-certificates。
• 时区不同步：容器用UTC，宿主机用CST，导致datetime.now()生成的时间戳错6小时。解决方案：挂载宿主机时区-v /etc/localtime:/etc/localtime:ro。

最稳妥的Dockerfile写法：

FROM python:3.9-slim RUN apt-get update && apt-get install -y tzdata && rm -rf /var/lib/apt/lists/* ENV TZ=Asia/Shanghai RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["python", "main.py"]

4.4 “ValueError: time data does not match format” —— 时间解析的100种死法

这个错误是K线清洗的头号杀手。根本原因：Pandas的to_datetime()默认用infer_datetime_format=True，但遇到混合格式（如2024-03-15和2024/03/15 09:30:00）会失败。

终极解决方案：永远显式指定格式，或用format='mixed'：

# 错误：依赖infer pd.to_datetime(df['time']) # 正确：混合格式 pd.to_datetime(df['time'], format='mixed') # 更正确：先统一字符串格式，再解析 df['time'] = df['time'].astype(str).str.replace('/', '-').str.replace(' ', 'T') pd.to_datetime(df['time'], errors='coerce')

errors='coerce'是保命参数——它把无法解析的时间转为NaT，后续可定位问题行。

4.5 “Dataframe is empty after conversion” —— 空数据的幽灵

API返回{"code":0,"data":[]}很常见，但你的代码若没检查len(klines)==0，直接pd.DataFrame([])会生成空DataFrame，后续df['close'].mean()报KeyError。

防御式写法：

def robust_kline_fetch(symbol: str, start: str, end: str) -> pd.DataFrame: resp = requests.get(f"https://api.xxx.com/kline?symbol={symbol}&start={start}&end={end}") resp.raise_for_status() data = resp.json() if not data.get('data', {}).get('klines'): # 或根据实际结构调整 logger.warning(f"No kline data for {symbol} from {start} to {end}") return pd.DataFrame(columns=['open','high','low','close','volume','time']) df, _ = generate_kline_dataframe(data, symbol) return df

返回空DataFrame时，必须带完整列名，否则下游concat()会报错。

5. 进阶实践：构建你的K线数据质量防火墙

5.1 自动化Schema监控：让API变更无所遁形

API字段变更无声无息，等你发现时策略已跑偏一周。我的方案是：每日定时抓取各API的schema样本，用diff算法告警。

实现步骤：

1. 对每个API端点，构造最小请求（如symbol=600519.SH&period=1d&count=1）
2. 保存响应结构到schema_history/20240315_tushare_daily.json
3. 用jsondiff库对比昨日schema：

from jsondiff import diff with open('schema_history/20240314.json') as f: old = json.load(f) with open('schema_history/20240315.json') as f: new = json.load(f) changes = diff(old, new) if changes: send_alert(f"API schema changed: {changes}")

我设置企业微信机器人，一旦检测到new['data']['klines'][0].keys()新增'pre_close_adj'或删除'amount'，立刻推送。

5.2 数据质量看板：用Grafana监控每一根K线

把quality_report指标写入InfluxDB，用Grafana看板监控：

• valid_ratio趋势图（目标>99.5%）
• invalid_reasons饼图（快速定位高频问题）
• processing_time_secP95延迟（超过5秒告警）

看板截图里，我最关注的是“停牌日填充率”——如果某只股票is_suspended==True的K线占比突增，说明它可能进入重大事项停牌，需人工核查。

5.3 回测沙盒：在真实数据上验证清洗效果

清洗再完美，不经过回测检验都是纸上谈兵。我的沙盒流程：

1. 用清洗后的DataFrame生成backtrader数据源
2. 运行一个极简策略（如“收盘价上穿5日均线买入”）
3. 对比清洗前后策略收益曲线
4. 若差异>0.5%，启动df.compare()定位哪根K线导致偏差

曾发现某平台日线数据中，2023-10-09（国庆后首个交易日）的open被错误填充为0，导致策略在该日空仓，损失一个涨停板。清洗模块的is_price_nonzero校验成功捕获此问题。

5.4 开源工具推荐：别重复造轮子，但要懂轮子怎么坏

• Akshare：国内最全免费金融数据源，但需注意其stock_zh_a_hist返回的DataFrame已做基础清洗，字段名统一，适合入门。
• baostock：券商级数据，login()后调用query_history_k_data_plus()，返回pd.DataFrame，但需自行处理peTTM等非K线字段。
• yfinance：美股首选，Ticker('AAPL').history(period="1mo")返回即用DataFrame，时区自动处理。

但记住：所有开源工具都只解决“搬运”，不解决“质检”。它们的README里不会写“本数据未校验价格逻辑有效性”，而这恰恰是你的护城河。

6. 我的血泪经验：三条铁律，保住你的策略性命

第一条铁律：永远在DataFrame里留一列source_api。
我见过太多团队，把Tushare、聚宽、akshare的数据concat()在一起，结果因volume单位不同（Tushare是“手”，聚宽是“股”），导致资金管理模块算错仓位。加一列source_api='tushare_v2'，后续groupby('source_api')就能隔离问题。

第二条铁律：清洗代码必须和策略代码部署在同一环境。
曾有个项目，清洗脚本用Python 3.8，策略回测用3.10，pd.Timestamp在3.10中默认纳秒精度，3.8中是微秒，导致df.index[0]在两环境差1000倍，回测结果天壤之别。现在我们用Docker Compose统一环境。

第三条铁律：第一次接入新API，先跑1000根K线的手动审计。
随机抽100根，用Excel打开，肉眼检查