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(fAuth 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: fBearer {token},Content-Type: application/json}params = {symbol: symbol, # 股票代码，如 600000.SHstart_date: start_date, # 格式 YYYY-MM-DDend_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-01date_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(fParse 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(fSaved {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(fNo new data for {symbol}. Last downloaded: {last_date})returnprint(fDownloading {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(fError 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(fAppended {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封禁。你更常用哪种写法？评论区交流 是倾向于SQLite本地存储以便离线分析，还是直接写入PostgreSQL以便多服务共享？或者你有更高效的并发下载方案？欢迎在评论区分享你的实战经验，一起打磨这套齐鲁证券同花顺下载的健壮性。