Python 小工具实战：用问财接口搭建金融数据采集器

前言构建金融数据采集链路时，团队常常在两条路之间反复横跳：要么基于 Tushare / akshare 等开源数据源自行拼接指标，开发成本高且覆盖维度受限；要么采购商业数据终端，预算压力显著。事实上，同花顺旗下的 i 问财（iwencai.com） 提供了一个被低估的结构化入口——它将自然语言问题翻译为内部查询计划，并返回标准化的多维数据。对工程侧而言，这相当于一个"托管的金融查询引擎"。本文的目标是：用 Python 将这一能力封装为可复用、可批量、具备基本健壮性的本地采集工具。我们先在基础版跑通请求链路，再在升级版补齐代理分散、失败重试与并发控制等工程要素。一、工具核心功能与准备工作1. 核心功能该工具围绕三个工程目标设计：自然语言到结构化数据的映射：将一句中文查询转换为字段对齐的数据行，规避自行实现筛选逻辑的成本。分页批量拉取：通过翻页迭代将满足条件的结果集完整导出，而非仅取首页。采集健壮性：以代理分散出口、指数退避重试等手段降低长任务被限流的概率（升级版重点）。2. 环境准备Python 版本：建议 3.10+（使用 from __future__ import annotations 与现代类型标注）。依赖安装：

代理资源（可选）：当采集频率提升、单机出口 IP 触发风控时，可引入代理 IP 池做出口分散。主流厂商（如亿牛云）提供隧道代理形态——即一个固定入口、由服务端自动轮换出口 IP，省去自建代理池的运维开销。本文升级版以该形态为例。接口机理说明：问财的查询走 POST /question/query，但接口要求请求携带由首页下发的 hexin-v 鉴权 cookie。该令牌具备时效性与会话绑定特征，因此正确做法是先用 Session 访问首页写入 cookie，再携带同一会话发起查询；硬编码令牌会导致 403。此外，响应体为嵌套 JSON（answer.components[].data.datas），字段名即前端展示的列名，且可能随版本演进，落库前建议做字段白名单校验。二、代码拆解与实现（附完整代码）1. 导入需要的库

2. 核心功能函数（采集的关键）（1）建立会话并获取鉴权令牌首页响应在 Set-Cookie 中写入 hexin-v，用 Session 自动管理后续请求的头域。

def bootstrap_session() -> requests.Session:
    """访问问财首页，建立持有 hexin-v 令牌的会话。

    Returns:
        已注入鉴权 cookie 的 requests.Session 实例。
    """
    session = requests.Session()
    session.headers.update({
        "User-Agent": (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/124.0 Safari/537.36"
        ),
        "Referer": "https://www.iwencai.com/",
    })
    session.get("https://www.iwencai.com/", timeout=10)
    return session

（2）执行查询并解析响应向 question/query 发送 POST，将多层 JSON 投影为扁平数据行。

def ask_wencai(session: requests.Session, question: str, perpage: int = 20) -> list[dict[str, Any]]:
    """发起一次自然语言查询并提取数据行。

    Args:
        session: 已初始化的会话对象。
        question: 自然语言问题，例如 "昨日涨停的非 ST 股票"。
        perpage: 单页返回条数。

    Returns:
        每条数据为一个字典，键为字段名，值为对应内容。

    Raises:
        requests.HTTPError: 当 HTTP 状态码为非 2xx。
        KeyError: 当响应结构不符合预期（接口变更）。
    """
    params = {
        "question": question,
        "perpage": perpage,
        "page": 1,
        "secondary_intent": "query",
        "log_info": "{}",
    }
    resp = session.post("https://www.iwencai.com/question/query", params=params, timeout=15)
    resp.raise_for_status()
    payload = resp.json()
    components = payload["answer"]["components"]
    data_block = next(c["data"] for c in components if "data" in c)
    return data_block.get("datas", [])

（3）分页迭代器以生成器抽象翻页，便于上层按需消费且避免一次性占用内存。

def iter_pages(
    session: requests.Session,
    question: str,
    perpage: int = 20,
    max_page: int = 5,
) -> Iterator[dict[str, Any]]:
    """按页遍历问财结果，逐条产出数据行。

    Args:
        session: 已初始化的会话对象。
        question: 自然语言问题。
        perpage: 单页条数。
        max_page: 最大翻页数，作为安全上限防止无限请求。

    Yields:
        单条数据字典。
    """
    for page in range(1, max_page + 1):
        params = {
            "question": question,
            "perpage": perpage,
            "page": page,
            "secondary_intent": "query",
            "log_info": "{}",
        }
        resp = session.post("https://www.iwencai.com/question/query", params=params, timeout=15)
        resp.raise_for_status()
        block = next(c["data"] for c in resp.json()["answer"]["components"] if "data" in c)
        rows = block.get("datas", [])
        if not rows:
            break
        yield from rows
        time.sleep(1)  # 基础版无代理，加入请求间隔以降低风控触发概率

3. 主程序入口

三、升级版金融数据采集器：代理分散与失败自愈3.1 功能升级说明基础版在低频场景下可用，但将其投入持续运行会暴露两个工程问题：出口 IP 信誉约束：目标站基于出口 IP 做频控，单机长时高频请求会触发限流甚至校验挑战。引入代理 IP 池可将流量分散到多个出口，使单次请求在统计上更接近独立用户。瞬态故障处理：网络抖动或服务端限流属常态，需以指数退避重试吸收，而非直接失败。升级版的工程要点：将代理、重试、分页收敛为可配置组件；以隧道代理形态实现出口自动轮换（以亿牛云隧道代理为例，其提供固定入口、由服务端轮询分配出口 IP）；通过 logging 替代 print 便于接入日志系统。3.2 完整代码实现（含代理与重试）

"""问财金融数据采集器 · 升级版（代理分散 + 指数退避重试）。"""

HOME = "https://www.iwencai.com"

QUERY = f"{HOME}/question/query"

def __init__(self, cfg: CollectorConfig) -> None:

self._cfg = cfg

self._proxies = {"http": cfg.proxy_url, "https": cfg.proxy_url}

self._session = requests.Session()

self._session.headers.update({

"User-Agent": (

"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "

"AppleWebKit/537.36 (KHTML, like Gecko) "

"Chrome/124.0 Safari/537.36"

),

"Referer": self.HOME + "/",

})

def collect(self, question: str, max_page: int = 5) -> list[dict[str, Any]]:

"""采集指定问题前若干页的全部数据。

Args:

question: 自然语言问题。

max_page: 最大翻页数。

Returns:

汇总后的数据行列表。

"""

self._session.get(self.HOME, proxies=self._proxies, timeout=10)

results: list[dict[str, Any]] = []

for page in range(1, max_page + 1):

rows = self._fetch_page(question, page)

if not rows:

break

results.extend(rows)

logger.info("采集完成，共 %d 条", len(results))

return results

def _fetch_page(self, question: str, page: int) -> list[dict[str, Any]]:

"""单页抓取，含指数退避重试。"""

params = {

"question": question,

"perpage": self._cfg.perpage,

"page": page,

"secondary_intent": "query",

"log_info": "{}",

}

for attempt in range(1, self._cfg.max_retries + 1):

try:

resp = self._session.post(

self.QUERY, params=params,

proxies=self._proxies, timeout=15,

)

if resp.status_code == 200:

block = next(c["data"] for c in resp.json()["answer"]["components"] if "data" in c)

return block.get("datas", [])

logger.warning("第 %d 次重试：HTTP %d", attempt, resp.status_code)

except requests.RequestException as exc:

logger.warning("第 %d 次重试：%s", attempt, exc)

time.sleep(self._cfg.backoff_base ** attempt)

logger.error("页码 %d 在重试后仍失败", page)

return []

if __name__ == "__main__":

cfg = CollectorConfig(proxy_url="http://<user>:<pass>@tunnel.ipidea.cn:2333")

collector = WencaiCollector(cfg)

data = collector.collect("北向资金连续三日增持的白酒股", max_page=3)

print(f"升级版共采集 {len(data)} 条")

工程收益：出口分散：代理将请求流量打散到多个出口 IP，缓解单 IP 频控；隧道形态下出口轮换由服务端完成，客户端零维护。故障吸收：指数退避重试吸收瞬态限流与网络抖动，提升长任务成功率。可观测与可扩展：logging 便于接入集中日志；CollectorConfig 将参数收敛，多实例并发即可线性扩容吞吐（注意控制单实例请求节奏，避免对目标站造成过大压力）。小结"基础版跑通链路 → 升级版补齐健壮性"的两步法，让我们以不到两百行代码获得一个可持续运行的金融数据采集器。核心难点不在解析，而在稳定性工程：鉴权令牌的生命周期管理、出口 IP 的信誉约束、以及瞬态故障的重试策略，才是决定采集系统能否长跑的关键。