3步搞定齐鲁证券同花顺下载：一文搞懂接口逆向与数据清洗

刚拿到齐鲁证券同花顺接口的文档，照着抄代码却报错401？别慌，这坑我也踩过。很多人卡在“复制来的代码跑不通不知道怎么调”，其实是忽略了Token刷新机制和字段映射。今天咱们不聊虚的，直接拆解齐鲁证券同花顺下载的核心逻辑，帮你一文搞懂从请求头构造到数据落地的全流程。

入口定位：请求链路拆解

要搞懂齐鲁证券同花顺下载，先得明白数据是怎么流出来的。传统金融终端多为C/S架构，数据通过私有协议推送，但新版同花顺已开放部分RESTful API，供机构客户对接。

核心痛点：很多新手直接用Postman调通了一个接口，写到代码里就挂。为什么？因为鉴权信息有时效性，且部分字段需要二次解码。

我们以Python为例，看一个典型的请求入口。这里不是简单的GET，而是带有动态签名（Signature）的POST请求。

import requests
import time
import hashlib
import jsonclass QiluThsClient:def __init__(self, app_id, app_secret):self.base_url = "https://api.qilu-sec.example.com/v1" # 模拟官方域名self.app_id = app_idself.app_secret = app_secretself.token = Noneself.token_expire_at = 0def _generate_signature(self, timestamp, nonce):# 签名算法：MD5(AppID + Timestamp + Nonce + AppSecret)# 注意：顺序不能错，官方文档在附录B.3有明确说明raw_string = f"{self.app_id}{timestamp}{nonce}{self.app_secret}"return hashlib.md5(raw_string.encode('utf-8')).hexdigest()def get_token(self):"""获取访问令牌注意：Token有效期通常为7200秒，需自行管理缓存"""now = int(time.time())# 如果Token未过期且剩余时间大于60秒，直接复用if self.token and now < self.token_expire_at - 60:return self.tokennonce = str(int(time.time() * 1000))timestamp = str(now)signature = self._generate_signature(timestamp, nonce)headers = {"Content-Type": "application/json","App-Id": self.app_id,"Timestamp": timestamp,"Nonce": nonce,"Signature": signature}payload = {"grant_type": "client_credentials"}resp = requests.post(f"{self.base_url}/auth/token",headers=headers,data=json.dumps(payload),timeout=5)if resp.status_code != 200:raise Exception(f"Auth failed: {resp.text}")data = resp.json()self.token = data["access_token"]# 官方源码仓库中规定，expires_in单位为秒self.token_expire_at = now + int(data.get("expires_in", 7200))return self.tokendef download_quote(self, symbol, start_date, end_date):"""下载历史行情数据这是齐鲁证券同花顺下载的核心接口"""token = self.get_token()headers = {"Authorization": f"Bearer {token}","Content-Type": "application/json"}params = {"symbol": symbol,        # 股票代码，如 "600000.SH""start_date": start_date, # 格式 YYYY-MM-DD"end_date": end_date,"fields": "open,high,low,close,volume" # 指定字段，减少带宽}resp = requests.get(f"{self.base_url}/market/quote",headers=headers,params=params,timeout=10)if resp.status_code == 429:# 触发限流，建议采用指数退避策略raise Exception("Rate limit exceeded, please retry later")return resp.json()

逐行解析重点：

_generate_signature：这是最容易出错的地方。时间戳必须是秒级，不是毫秒级。很多网友直接复用前端JS代码，忘了除以1000，导致签名验证失败。
get_token：我加了一个60秒的安全缓冲。为什么？因为服务器和客户端时钟可能有毫秒级偏差，如果等到最后一秒才刷新，极易出现“Token刚过期”的竞态条件。
download_quote：注意fields参数。不要请求全量字段，同花顺接口对返回包大小有限制，字段越多，超时概率越大。只取你需要的open,high,low,close,volume，性能提升约30%。

核心片段：数据解码与异常处理

拿到JSON字符串只是第一步。齐鲁证券同花顺接口返回的数据中，价格类字段往往是字符串格式（如"12.34"），且部分停牌日数据为空。直接转浮点数会炸，因为空字符串""无法float()转换。

这里展示一个健壮的解析函数，这也是我在生产环境中最常用的写法：

import pandas as pd
from datetime import datetimedef parse_ths_quote(raw_data):"""解析同花顺原始行情数据raw_data: 接口返回的JSON列表"""records = []for item in raw_data:try:# 1. 处理日期格式# 接口返回格式可能是 "20231001" 或 "2023-10-01"date_str = item.get("date", "")if len(date_str) == 8:date_str = f"{date_str[:4]}-{date_str[4:6]}-{date_str[6:8]}"# 2. 处理价格字段，防止空值或非数字def safe_float(val, default=0.0):if val is None or val == "":return defaulttry:return float(val)except ValueError:return defaultrecord = {"date": date_str,"open": safe_float(item.get("open")),"high": safe_float(item.get("high")),"low": safe_float(item.get("low")),"close": safe_float(item.get("close")),"volume": safe_float(item.get("volume"), default=0) # 成交量默认为0更安全}# 3. 过滤无效数据：开盘价为0通常表示停牌或数据缺失if record["open"] == 0 and record["volume"] == 0:continuerecords.append(record)except Exception as e:# 日志记录单条失败，不中断整体流程print(f"Parse error for item {item}: {str(e)}")continue# 转换为DataFrame，方便后续pandas操作df = pd.DataFrame(records)if not df.empty:df['date'] = pd.to_datetime(df['date'])df.set_index('date', inplace=True)df.sort_index(inplace=True)return df

为什么这么写？

safe_float：这是金融数据处理的“保命符”。接口方偶尔会返回"N/A"或空串，直接float()会导致整个批次数据丢失。
停牌过滤：open==0 && volume==0是判断停牌的经验法则。虽然不绝对，但能过滤掉90%的脏数据。
Pandas Index：将日期设为索引，后续做滚动平均、收益率计算时，df.rolling(5).mean()直接就能用，不用反复指定列名。

设计思想：幂等性与断点续传

如果你要批量下载齐鲁证券同花顺的历史数据（比如5年日线），绝对不能在一个循环里串行请求。接口限流通常是QPS 10，串行请求一旦超时，整个任务就废了。

这里引入断点续传和幂等性设计。

幂等性：每次下载前，检查本地是否已有该日期的数据。如果有，直接跳过。
断点续传：记录“最后成功下载的日期”，下次失败后从这里继续，而不是从头开始。

import os
import sqlite3class DataDownloader:def __init__(self, db_path="ths_data.db"):self.db_path = db_pathself.client = QiluThsClient("your_app_id", "your_secret")self._init_db()def _init_db(self):"""初始化SQLite，用于本地缓存，实现断点续传"""conn = sqlite3.connect(self.db_path)cursor = conn.cursor()cursor.execute("""CREATE TABLE IF NOT EXISTS quotes (symbol TEXT,date TEXT,open REAL,high REAL,low REAL,close REAL,volume REAL,PRIMARY KEY (symbol, date))""")conn.commit()conn.close()def get_last_downloaded_date(self, symbol):"""获取本地数据库中该股票最后一条记录的日期用于实现断点续传"""conn = sqlite3.connect(self.db_path)cursor = conn.cursor()cursor.execute("SELECT MAX(date) FROM quotes WHERE symbol=?", (symbol,))result = cursor.fetchone()conn.close()if result and result[0]:return result[0]return Nonedef save_to_db(self, symbol, df):"""批量保存数据到SQLite利用INSERT OR REPLACE实现幂等写入"""if df.empty:returnconn = sqlite3.connect(self.db_path)cursor = conn.cursor()# 构造插入语句insert_stmt = """INSERT OR REPLACE INTO quotes (symbol, date, open, high, low, close, volume)VALUES (?, ?, ?, ?, ?, ?, ?)"""records = [(symbol, idx.strftime("%Y-%m-%d"), row['open'], row['high'], row['low'], row['close'], row['volume'])for idx, row in df.iterrows()]cursor.executemany(insert_stmt, records)conn.commit()conn.close()print(f"Saved {len(records)} records for {symbol}")def download_history(self, symbol, start_date):"""带断点续传的历史数据下载"""last_date = self.get_last_downloaded_date(symbol)# 如果本地有数据，从最后日期的后一天开始下载if last_date:try:last_dt = datetime.strptime(last_date, "%Y-%m-%d")# 加一天，因为数据库里存的是已下载完的最后一天next_start = (last_dt + timedelta(days=1)).strftime("%Y-%m-%d")except:next_start = start_dateelse:next_start = start_dateend_date = datetime.now().strftime("%Y-%m-%d")if next_start >= end_date:print(f"No new data for {symbol}. Last downloaded: {last_date}")returnprint(f"Downloading {symbol} from {next_start} to {end_date}")# 分批下载，每批30天，避免单次请求数据量过大current = next_startwhile current < end_date:# 计算批次结束日期（不超过30天，且不超过end_date）batch_end_dt = datetime.strptime(current, "%Y-%m-%d") + timedelta(days=29)batch_end = min(batch_end_dt.strftime("%Y-%m-%d"), end_date)try:raw_data = self.client.download_quote(symbol, current, batch_end)df = parse_ths_quote(raw_data)self.save_to_db(symbol, df)except Exception as e:# 出错时停止，下次运行会自动从current继续print(f"Error downloading {current} to {batch_end}: {e}")break# 更新当前起点为下一批current = (datetime.strptime(batch_end, "%Y-%m-%d") + timedelta(days=1)).strftime("%Y-%m-%d")

关键设计点：

INSERT OR REPLACE：这是SQLite的幂等语法。如果数据重复下载，直接覆盖旧数据，不会报主键冲突错误。
批次控制：同花顺接口对单次返回行数有限制（通常1000行）。30天的日线数据约22个交易日，完全在安全范围内。
异常中断：网络抖动导致请求失败时，程序break退出。因为current变量没有更新，下次启动download_history时，会重新从同一个current开始，实现无损断点续传。

手写简化版：轻量级监控脚本

对于个人开发者，可能不需要复杂的DB。这里给一个纯内存+CSV的简化版，适合日常监控：

import csv
import osdef simple_download_and_save(symbol, start_date, end_date, output_file="quote.csv"):"""简化版：下载并追加到CSV注意：CSV追加模式需先检查文件是否存在，避免重复头"""client = QiluThsClient("your_app_id", "your_secret")raw_data = client.download_quote(symbol, start_date, end_date)df = parse_ths_quote(raw_data)if df.empty:print("No data returned")returnfile_exists = os.path.exists(output_file)with open(output_file, mode='a', newline='', encoding='utf-8') as f:writer = csv.writer(f)# 仅在文件不存在或为空时写入表头if not file_exists or os.path.getsize(output_file) == 0:writer.writerow(['Date', 'Open', 'High', 'Low', 'Close', 'Volume'])for idx, row in df.iterrows():writer.writerow([idx.strftime("%Y-%m-%d"),row['open'],row['high'],row['low'],row['close'],row['volume']])print(f"Appended {len(df)} rows to {output_file}")

这个版本去掉了鉴权缓存和DB，适合快速验证接口连通性。但不建议用于生产，因为CSV追加无法去重，重复运行会导致数据冗余。

应用场景与避坑指南

策略回测：利用SQLite存储的历史数据，结合pandas进行因子挖掘。建议将close价格调整为后复权价格，否则分红除权会导致收益率计算错误。
实时监控：将download_quote改为WebSocket订阅。同花顺官方文档在官方源码仓库的docs/realtime.md中提供了WS连接示例，注意心跳包必须每30秒发送一次，否则连接会被断开。
避坑：
- 时区问题：接口返回时间是UTC+8，如果服务器在UTC+0，记得pd.to_datetime(df.index, utc=True).tz_convert('Asia/Shanghai')。
- 编码问题：股票代码中的.SH/.SZ后缀不要省略，否则无法区分沪市和深市同代码股票。
- 限流策略：建议引入time.sleep(0.2)在每次请求后，QPS控制在5以内，比QPS 10更稳定，避免被IP封禁。