quant_data · 数据工具箱 2026-09-27

from datetime import datetime # 导入 datetime：Python 自带，用来处理时间 import numpy as np # 导入 numpy：数值计算库（我们叫它"发动机"） import pandas as pd # 导入 pandas：表格处理库（我们叫它"工作台"）

def get_kline(symbol="600519", start_date="20260101", end_date=None): """ 拉取 A 股日线，返回一张表格，列名统一为： open（开盘）high（最高）low（最低）close（收盘）volume（成交量） 数据源自动兜底：akshare -> baostock -> 腾讯，谁通就用谁。 symbol 是股票代码，默认贵州茅台 600519。 """ import akshare as ak # akshare：免费 A 股数据源 import baostock as bs # baostock：备用数据源 import requests # requests：用来访问网络接口

# 如果调用时没传 end_date，就用今天的日期 if end_date is None: end_date = datetime.now().strftime("%Y%m%d") # datetime.now()：拿到当前这一刻的时间 # strftime("%Y%m%d")：把时间格式化成"年月日"连写，如 20260814 # baostock 要的是带横杠的日期，把 "20260101" 拆成 "2026-01-01" end_dash = f"{end_date[:4]}-{end_date[4:6]}-{end_date[6:]}" start_dash = f"{start_date[:4]}-{start_date[4:6]}-{start_date[6:]}" # f-string（新语法）：字符串前加 f，就能用 {变量} 把值塞进字符串里 # 字符串切片：end_date[:4] 取前 4 个字符（年份），[4:6] 取第 5、6 个（月份）

# ---- 数据源 1：akshare（东财，免费、最常用）---- try: # try："尝试"运行下面几行；万一报错不崩溃，会跳到 except raw = ak.stock_zh_a_hist( # 调用 akshare 的"A 股历史行情"接口 symbol=symbol, # 股票代码 period="daily", # 周期：daily 表示日线 start_date=start_date, # 开始日期 end_date=end_date, # 结束日期 adjust="qfq", # 前复权：把分红除权的影响折算回价格里 ) # 判断：确实拿到了数据才继续（raw 不是空、行数大于 0） if raw is not None and len(raw) > 0: # 把 akshare 返回的中文列，整理成统一的英文列 df = pd.DataFrame({ # DataFrame：pandas 的"一张表" "date": pd.to_datetime(raw["日期"]), # 日期列转成时间格式 "open": raw["开盘"].astype(float), # 开盘价，转成小数 "high": raw["最高"].astype(float), # 最高价 "low": raw["最低"].astype(float), # 最低价 "close": raw["收盘"].astype(float), # 收盘价 "volume": raw["成交量"].astype(float), # 成交量 }) # 把"日期"这一列设为表格的索引（以后就能按日期切片、排序） df = df.set_index("date") print(f"[数据源] akshare 获取成功，共 {len(df)} 条") # 打印来源，方便排错 return df # return：把表格返回给调用这个函数的地方 except Exception as e: # 如果上面 try 里报错了，就跳到这一行 print(f"[数据源] akshare 不可用（{type(e).__name__}），尝试 baostock...") # type(e).__name__：把错误类型变成名字打印出来，方便知道错在哪

# ---- 数据源 2：baostock（稳定，数据截至上一个交易日）---- try: # baostock 的股票代码带市场前缀：6 开头是上海（sh），其他是深圳（sz） bs_code = ("sh." if symbol.startswith("6") else "sz.") + symbol bs.login() # baostock 要先登录（免费的，不用注册密码） rs = bs.query_history_k_data_plus( # 查询历史日线 bs_code, # 带前缀的股票代码 "date,open,high,low,close,volume", # 要哪些字段 start_date=start_dash, # 开始日期（带横杠格式） end_date=end_dash, # 结束日期 frequency="d", # 频率：d = 日线 adjustflag="2", # 2 = 前复权 ) rows = [] # 建一个空列表，准备装每一行数据 while rs.error_code == "0" and rs.next(): # 循环：一页一页拿数据 rows.append(rs.get_row_data()) # 把这一行装进列表 bs.logout() # 用完退出登录 if rows: # 如果确实拿到了数据 # 把列表变成表格，并指定列名 df = pd.DataFrame(rows, columns=["date", "open", "high", "low", "close", "volume"]) for c in ("open", "high", "low", "close", "volume"): # 遍历这 5 个价格列 df[c] = pd.to_numeric(df[c]) # 字符串转成数字 df["date"] = pd.to_datetime(df["date"]) # 日期列转成时间格式 df = df.set_index("date") # 日期设为索引 print(f"[数据源] baostock 获取成功，共 {len(df)} 条") return df # 返回表格 except Exception as e: print(f"[数据源] baostock 不可用（{type(e).__name__}），尝试腾讯...")

# ---- 数据源 3：腾讯 ifzq（前复权，最多 1000 条）---- try: ten_code = ("sh" if symbol.startswith("6") else "sz") + symbol # 腾讯格式代码 url = "https://web.ifzq.gtimg.cn/appstock/app/fqkline/get" # 腾讯 K 线接口 r = requests.get(url, params={"param": f"{ten_code},day,,,1000,qfq"}, timeout=15) # requests.get：向接口发请求；params：接口要的参数；timeout：最多等 15 秒 kline = r.json()["data"][ten_code] # 把返回的 JSON 数据取出来 key = "qfqday" if "qfqday" in kline else "day" # 有前复权数据就用它 rows = kline[key] # 取出行情列表 # 注意：腾讯返回的顺序是 日期/开盘/收盘/最高/最低/成交量 df = pd.DataFrame(rows, columns=["date", "open", "close", "high", "low", "volume"]) for c in ("open", "close", "high", "low", "volume"): # 5 列都转成数字 df[c] = pd.to_numeric(df[c]) df["date"] = pd.to_datetime(df["date"]) # 日期转时间格式 df = df.set_index("date") # 日期设为索引 print(f"[数据源] 腾讯 ifzq 获取成功，共 {len(df)} 条") return df # 返回表格 except Exception as e: print(f"[数据源] 腾讯也不可用（{type(e).__name__}）") # 三个数据源都不行，主动报错提示检查网络 raise RuntimeError("所有 A 股数据源都不可用，请检查网络")

def get_kline_crypto(symbol="BTC/USDT", limit=1500): """ 拉取加密市场日线，列名统一为 open/high/low/close/volume。 数据源自动兜底：ccxt + OKX（上期注册的交易所）-> 币安公开行情。 symbol 是交易对，默认 BTC/USDT。 """ import ccxt # ccxt：一个库连接几百家交易所（上期介绍过） import requests # requests：访问网络接口

# ---- 数据源 1：ccxt + OKX（上期注册的 API，可选）---- try: # 如果 .env 里配了 Key（okx_setup.py 教过），就带上 Key 连接 import os # os：读取系统设置（环境变量） api_key = os.getenv("OKX_API_KEY", "").strip() # 读 API Key（没有就是空） secret = os.getenv("OKX_SECRET", "").strip() # 读 Secret password = os.getenv("OKX_PASSWORD", "").strip() # 读密码短语 # 构造 OKX 连接对象；三个值都有才带上 Key，否则就是匿名访问 if api_key and secret and password: # 三个都填了 ex = ccxt.okx({"apiKey": api_key, "secret": secret, "password": password, "enableRateLimit": True, "timeout": 30000}) else: # 没填 Key ex = ccxt.okx({"enableRateLimit": True, "timeout": 30000}) # enableRateLimit：自动控制请求频率，防止被限流 proxy = os.getenv("OKX_PROXY", "").strip() # 可选的代理地址 if proxy: # 如果配了代理 ex.proxies = {"http": proxy, "https": proxy} # 走代理访问 bars = ex.fetch_ohlcv(symbol, "1d", limit=limit) # 拉日线（1d=一天一根） if bars: # 拿到了数据 # fetch_ohlcv 返回的是列表套列表，转成表格 df = pd.DataFrame(bars, columns=["ts", "open", "high", "low", "close", "volume"]) df["date"] = pd.to_datetime(df["ts"], unit="ms") # 毫秒时间戳转日期 df = df.set_index("date") # 日期设为索引 df = df.drop(columns=["ts"]) # 把原始时间戳列删掉 print(f"[数据源] OKX(ccxt) 获取成功，共 {len(df)} 条") return df # 返回表格 except Exception as e: print(f"[数据源] OKX 不可用（{type(e).__name__}），尝试币安公开行情...")

# ---- 数据源 2：币安公开行情（无需注册，直连稳定）---- try: base = symbol.replace("/", "") # "BTC/USDT" 去掉斜杠变成 "BTCUSDT" url = "https://data-api.binance.vision/api/v3/klines" # 币安公开行情接口 r = requests.get(url, params={"symbol": base, "interval": "1d", "limit": limit}, timeout=20) bars = r.json() # 解析返回的 JSON if isinstance(bars, list) and bars: # isinstance：判断是不是列表且非空 # 币安返回 12 列，我们只取前 6 列（时间/开/高/低/收/量） df = pd.DataFrame(bars, columns=[ "ts", "open", "high", "low", "close", "volume", "close_ts", "quote_vol", "n", "taker_buy", "taker_buy_quote", "ignore", ]) df = df[["ts", "open", "high", "low", "close", "volume"]] # 只留需要的列 for c in ("open", "high", "low", "close", "volume"): # 价格列转数字 df[c] = pd.to_numeric(df[c]) df["date"] = pd.to_datetime(df["ts"], unit="ms") # 时间戳转日期 df = df.set_index("date") # 日期设为索引 df = df.drop(columns=["ts"]) # 删掉时间戳列 print(f"[数据源] 币安公开行情 获取成功，共 {len(df)} 条") return df # 返回表格 except Exception as e: print(f"[数据源] 币安也不可用（{type(e).__name__}）") raise RuntimeError("所有加密数据源都不可用，请检查网络")