企业官网建设流程全解析

简介：面向Python爬虫学习者和金融数据采集爱好者，这份资源以雪球网这一真实投资社区为目标，专注于抓取大V组合的调仓记录，可用于跟踪投资动态、分析市场趋势、复盘大V操作逻辑。压缩包共2个文件，均为.py脚本，大小仅10KB；其中v6.3正式版采用4线程并发抓取以提升效率，v4早期版则结构更直接，适合先读它再对照升级版。已有382人浏览学习。资源描述详述了从Requests构造HTTP请求、BeautifulSoup解析页面、通过选择器定位数据、再到存储落盘的完整流程，也提及多线程加速和遵守robots协议等实操要点；抓取结果可进一步用于趋势分析、投资策略研究和风险评估等场景。对比两个不同时期的版本，可以清晰看出爬虫在调度、容错和并发上的演进，从而快速掌握股票类数据抓取的基本套路；对于想入门分析型爬虫的开发者而言，这是一份小巧而完整的学习样本。包体小巧，下载后即可快速查看全部源码。很适合作为爬虫课程作业或投资数据采集的起步参考。

1. 调仓记录爬虫：把人工盯盘变成可回放的数据流

炒股的人都有过这种体验：打开股票社区App，直奔几个心仪的组合页，翻今天有没有调仓，买没买，卖没卖，仓位又变多少。如果只关注一两个组合还行，一旦关注了十来个，手动刷新就变成了负担，而且很容易错过盘中临时调仓。这个标题给的方案，就是用Python爬虫把股票社区里这些公开组合的调仓记录批量抓取下来，存进数据库，让盯盘从“人肉刷新”变成“定时轮询”。它抓的不是行情K线，而是一笔笔发生在组合里的交易动作，正好适合用来复盘高手的操作思路。标题里的“两个版本”也值得展开说：一个是轻量同步版，用来快速跑通逻辑；另一个是异步并发版，用来在组合数量变大时不拖后腿。无论你是刚学会requests的入门者，还是已经在爬虫上踩过坑的开发者，这套链路都能直接落地。

2. 先弄懂调仓数据从哪来：接口识别、字段约定与最小抓取闭环

在写第一行代码之前，得先搞清楚目标数据长什么样。这个环节做得扎实，后面可以少改一天代码。

2.1 调仓记录和持仓快照是两种数据，别混着抓

股票社区的组合功能，对外暴露的数据可以分成两类。第一类是当前持仓快照，它回答的是“这个组合此刻拿着哪些股票、各占多少比例”，是一个静态切片。第二类是调仓记录，它回答的是“这个组合过去每一次买入、卖出是在什么时候、什么价格、产生了什么仓位变化”，是一条时间序列。标题里要抓的是后者，但实际做数据对齐时，往往需要把两者合并使用：调仓记录告诉你操作历史，持仓快照告诉你当前结果。

一条完整的调仓记录，字段基本逃不出下面这些：

这里必须强调一个新手常见误区：不要上来就抓HTML页面。组合页面的调仓列表虽然可以用XPath抽取，但页面DOM只要改版，解析器立刻作废。更合理的路径是找到页面背后的JSON接口，因为接口返回的数据结构清晰固定，字段与表格里的列几乎一一对应，而且方便翻页。

2.2 用开发者工具把接口定位出来，顺便把签名参数看明白

定位接口不是玄学，而是固定的排查步骤。我一般这样做：

1. 用浏览器打开一个组合详情页，按F12进入开发者工具。
2. 切到Network面板，筛选类型为Fetch/XHR。
3. 刷新页面，或者点击调仓历史的翻页按钮。
4. 在请求列表里找一个返回内容为JSON、URL中带portfolio、transaction、history、trade之类的请求。
5. 点开该请求，分别看Headers和Payload，记录完整的URL、查询参数、请求头。

大多数组合调仓接口的查询参数都很薄，常见就这几个：

有些接口会在Query里多一个签名参数，比如sig=...或token=...。看到这个先不要慌，你可以把整个URL复制出来，放在浏览器地址栏里直接访问。如果能返回同样的JSON，说明签名参数是固定的或者绑定Cookie的，直接在爬虫里带上即可。如果浏览器里能访问、requests里却返回错误，最常见原因是你漏带了Cookie或Referer，优先检查这两个，而不是去纠结签名算法。大部分社区站的反爬重心放在登录态上，签名反而是次要防线。

2.3 返回结构两种常见形态与字段映射表

接口定位到之后，不要急着复制URL到代码里，先手动请求一次，把返回JSON的结构看清楚。组合类接口常见的返回形态有两种。形态A是分页包装型，外层是响应码和data，data里带total和items：

{ "code": 0, "data": { "total": 56, "items": [ { "id": 1001, "stock_code": "STK001", "stock_name": "示例标的", "action": "buy", "price": 10.25, "position_ratio": 0.15, "happened_at": 1699999999 } ] } }

形态B是平铺数组型，data直接就是一个列表，每条记录可能是包装过的对象：

{ "data": [ { "change": { "stock_code": "STK002", "stock_name": "示例标的B", "action": "sell", "price": 20.10, "created_at": "2023-11-15 10:30:00" } } ] }

两种形态的字段名也可能不同，比如happened_at在另一个接口里叫created_at，position_ratio可能叫weight。所以不要只写一套解析代码打天下，建议在代码里做一个字段映射函数，把接口原始字段统一转换成你内部的命名，后边所有逻辑只依赖内部命名。

def map_record(raw, portfolio_id): """把接口原始字段统一成内部字段""" return { "portfolio_id": portfolio_id, "stock_code": raw.get("stock_code") or raw.get("symbol"), "stock_name": raw.get("stock_name") or raw.get("name"), "action": raw.get("action") or raw.get("type"), "price": normalize_price(raw.get("price")), "position_ratio": raw.get("position_ratio") or raw.get("weight"), "happened_at": raw.get("happened_at") or raw.get("created_at"), }

normalize_price的作用是把"--"、空字符串、None统统转成None，这样入库时不会报类型错误。

2.4 最小验证脚本：先拉一页调仓记录，确认JSON结构再写正式版

有了上面的字段映射，就可以写最小验证脚本了。这个脚本的目的不是抓全量数据，而是验证请求参数对不对、返回结构是不是预期中的那一种。

import requests session = requests.Session() # 从浏览器复制登录后的Cookie，填成字典 cookies = { "sessionid": "你的登录态", "other_cookie": "值", } headers = { "User-Agent": "Mozilla/5.0", "Referer": "https://stock.example.com/portfolio/123456", } # 接口地址以开发者工具里实际抓到的为准，这里只是占位 url = "https://stock.example.com/api/portfolio/123456/transactions" resp = session.get( url, params={"page": 1, "size": 20}, headers=headers, cookies=cookies, timeout=10, ) print(resp.status_code) print(resp.text[:3000])

这个代码块有几个关键点。session用来保持Cookie和连接池，后续翻页都复用同一个会话，能减少重复登录校验。Referer必须指向组合页，很多服务端会校验来源，不填或填错会被拒。timeout=10是给请求设了超时上限，不然网络抖动时脚本可能挂住几分钟。跑通之后，打印出来的JSON就是后面解析逻辑的依据。如果打印出来的是HTML，先检查Cookie是否过期、Referer是否填错，再考虑是不是触发了验证码。

3. 两个版本怎么拆：同步requests版与异步aiohttp版的分工与实现

标题里的“两个版本”，放到工程里通常不是两个功能完全不同的程序，而是一个业务逻辑、两个传输层实现。下面分别说怎么落地。

3.1 版本一：requests + Session，按组合ID顺序拉分页

同步版是最容易写、最容易调试的版本，适合组合数量在几十个以内、抓取频率低（比如半小时一次）的场景。它的核心是循环分页，把某个组合的调仓记录一页一页拉完。

import time import random def fetch_transactions(session, portfolio_id, max_pages=20, size=50): all_items = [] for page in range(1, max_pages + 1): params = {"page": page, "size": size} resp = session.get( "https://stock.example.com/api/portfolio/{}/transactions".format(portfolio_id), params=params, timeout=10, ) if resp.status_code != 200: break data = resp.json() items = data.get("data", {}).get("items", []) if not items: break all_items.extend(items) time.sleep(random.uniform(0.5, 1.5)) return all_items session = requests.Session() session.headers.update({ "User-Agent": "Mozilla/5.0", "Referer": "https://stock.example.com/portfolio/123456", }) records = fetch_transactions(session, portfolio_id=123456) print("共抓到", len(records), "条调仓记录")

这个实现里最值得说的不是请求本身，而是两个break条件。第一个break处理HTTP异常：状态码不是200就停，不给服务端继续施压。第二个break处理分页结束：items为空说明没有更多数据，继续翻页只会浪费请求。time.sleep(random.uniform(0.5, 1.5))定义了每页之间的等待时间，区间加抖动是为了避免固定节奏被限速策略识别。如果你跑下来没有触发任何风控，可以把区间缩到0.2到0.5秒；如果发现有验证码返回，就要反向把区间放大到2到3秒。

分页参数里有两个细节。第一，page有的接口从0开始，如果第一页是空就切换起始值。第二，size设置过大时，服务端可能会忽略，强制每页20条，所以代码里不要依赖返回条数等于size。

3.2 版本二：aiohttp + semaphore，并发抓多个组合

当关注的组合从几十个涨到几百个，同步版就会显得很长。假设一个组合平均翻3页，每页间隔1秒，200个组合光等待就要600秒，加上请求时间就超过10分钟。这时需要把同步版升级成异步版，用aiohttp的事件循环把等待时间重叠起来。

import asyncio import aiohttp async def fetch_one(session, portfolio_id, semaphore, retry=3): async with semaphore: for attempt in range(retry): try: url = "https://stock.example.com/api/portfolio/{}/transactions".format(portfolio_id) params = {"page": 1, "size": 50} async with session.get(url, params=params, timeout=10) as resp: if resp.status == 200: data = await resp.json() return portfolio_id, data.get("data", {}).get("items", []) except (asyncio.TimeoutError, aiohttp.ClientError): pass await asyncio.sleep(1 * (attempt + 1)) return portfolio_id, [] async def fetch_many(portfolio_ids, concurrency=10): semaphore = asyncio.Semaphore(concurrency) timeout = aiohttp.ClientTimeout(total=30) async with aiohttp.ClientSession(timeout=timeout) as session: tasks = [fetch_one(session, pid, semaphore) for pid in portfolio_ids] results = await asyncio.gather(*tasks) return dict(results) portfolio_ids = [123456, 234567, 345678] result_map = asyncio.run(fetch_many(portfolio_ids, concurrency=10))

讲两个参数。第一个是Semaphore(concurrency)，它把并发请求数量钉在10。没有这一行，你会在同一瞬间发出上百个请求，结果不是更快，而是更快触发风控。第二个是retry和退避等待，这里用1 * (attempt + 1)，第一次失败等1秒，第二次等2秒，第三次等3秒。对于网络爬虫，指数退避更常见，但组合接口的偶发超时用线性退避已经够了，指数退避反而会让重试窗口拖太久。

这段代码目前只抓了每个组合的第一页，正式场景需要把翻页逻辑也搬进来，可以用一个while page <= max_pages的循环包住session.get，判断items为空就退出。异步版和同步版最好共用同一个map_record和入库函数，只替换请求层，这样两个版本抓出来的数据结构保持一致。

3.3 两个版本共用的数据落地：SQLite去重与增量更新

两个版本抓回来的数据，最终要进同一个地方。我偏好SQLite，单文件部署，不用起服务，适合个人爬虫。表结构设计时，去重键是最关键的一个决策。

CREATE TABLE transactions ( portfolio_id INTEGER, stock_code TEXT, stock_name TEXT, action TEXT, price REAL, position_ratio REAL, happened_at INTEGER, created_at TEXT DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (portfolio_id, stock_code, happened_at, action) );

主键用了四个字段的组合：组合ID、股票代码、调仓时间、操作类型。为什么把action也放进去？因为同一个组合的同一只股票在同一个时间戳里，理论上有buy和sell两条记录，如果只看前三个字段，第二条会被错误地忽略。这在实际数据里不常见，但为了避免边缘情况，我习惯把action也纳入去重键。

写入方式选择INSERT OR IGNORE还是INSERT OR ON CONFLICT DO UPDATE？如果只做历史累积，用INSERT OR IGNORE就行了。但我推荐后者，因为它能处理接口偶发的重复推送：去重键冲突时，用新数据覆盖旧字段，保证库里始终是最新一次的内容。

import sqlite3 conn = sqlite3.connect("portfolio_trades.db") upsert_sql = """ INSERT INTO transactions (portfolio_id, stock_code, stock_name, action, price, position_ratio, happened_at) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(portfolio_id, stock_code, happened_at, action) DO UPDATE SET stock_name = excluded.stock_name, price = excluded.price, position_ratio = excluded.position_ratio """ for item in records: conn.execute(upsert_sql, ( item["portfolio_id"], item["stock_code"], item["stock_name"], item["action"], item.get("price"), item.get("position_ratio"), item["happened_at"], )) conn.commit()

代码里有一个容易踩的细节：访问item["price"]之前如果没确认字段存在，会在缺失时报KeyError。所以我特意用item.get("price")，让缺失字段变成None写入数据库，后面分析时再统一处理。入库前最好把字符串数字转成float，否则SQLite会把150.00存成"150.00"字符串，后续聚合计算时类型会混乱。

4. 参数和策略怎么调：请求头、频率控制、重试与Cookies过期

代码能跑通只是开始，真正决定这个爬虫能跑多久的是参数和策略。这一章把请求头、频率、重试和登录态维护一次说清楚。

4.1 必调的请求头与Cookie传入方式

组合调仓接口几乎都做来源校验，四个头部字段是必调的：

User-Agent不要只写一个。我一般准备5个不同浏览器的UA，在Session初始化时随机选一个，后续整个会话保持一致。随机UA能降低被简单地按UA黑名单拦截的概率，但它不是核心防御，核心还是Cookie和请求频率。Referer的正确性比UA更重要，很多接口会校验这个头，填错直接返回403。检查时优先看Referer，再检查Cookie。

Cookie的保存方式也要讲。不建议把Cookie硬编码在代码里。常见做法是把浏览器里复制出来的Cookie字符串存到一个配置文件里，脚本启动时读取并解析成字典。Cookie快过期时，接口的表现是返回200但内容是一段HTML登录页，或者返回401。针对这种情况，我会在启动时先请求一次固定的组合ID，校验响应Content-Type是不是JSON，不是就尽早退出，避免后续所有请求都在白做。

4.2 频率控制：sleep、随机抖动和动态限速

频率控制的目的是让请求节奏看起来像人，而不是被限速规则盯上。单一维度的固定sleep效果有限，需要配合随机抖动。

import time import random def polite_sleep(base=1.5, jitter=1.0): time.sleep(base + random.uniform(0, jitter))

这里的base是最小等待秒数，jitter是随机上浮范围。假设base=1.5、jitter=1.0，实际等待时间在1.5到2.5秒之间波动。同步版在每页之间调用polite_sleep()，异步版则要改成await asyncio.sleep(base + random.uniform(...))，因为time.sleep会阻塞事件循环，导致并发全部卡住。

如果触发了风控，光靠固定sleep不够，需要动态降级。我通常记录最近10次请求的状态，当非200响应连续出现3次时，把间隔临时拉大到10秒，并持续5分钟。这个“动态限速”策略比单纯把sleep设大要划算，因为正常情况下它保持低延迟，遇到风险又会及时退让。

4.3 重试与失败判断：状态码/JSON结构双重检查

爬虫的失败不总是以状态码表示。很多风控系统返回200但内容是验证页，或者返回JSON但业务code不是0。所以重试条件必须做双重检查。

import time def safe_get(session, url, params, retry=3): for attempt in range(retry): try: resp = session.get(url, params=params, timeout=10) if resp.status_code != 200: raise RuntimeError("status {}".format(resp.status_code)) json_data = resp.json() if "data" not in json_data: raise ValueError("no data field") items = json_data.get("data", {}).get("items", []) return items except Exception as e: wait = 2 ** attempt print("第{}次重试，原因：{}".format(attempt + 1, e)) time.sleep(wait) return []

注意重试等待用的是2 ** attempt，第一次1秒，第二次2秒，第三次4秒。这个退避曲线在爬虫场景里很通用，既不会频繁重试，也不会等太久。另外不要对同一个请求无限重试，三次失败后返回空列表，由上层决定是告警还是放弃。打印错误日志时要把状态码、响应前200字符、参数里的组合ID都带上，方便事后定位。

4.4 Cookie过期检测：启动时先做一次健康检查

Cookie过期是这种带登录态的爬虫最常见的死法。防止办法是：在正式抓取前，用一个已知组合ID跑一次最小请求，用结果决定是否继续。

def health_check(session, probe_portfolio_id): url = "https://stock.example.com/api/portfolio/{}/transactions".format(probe_portfolio_id) try: resp = session.get(url, params={"page": 1, "size": 1}, timeout=10) content_type = resp.headers.get("Content-Type", "") if resp.status_code != 200 or "json" not in content_type: return False, resp.status_code, content_type data = resp.json() if "data" not in data: return False, resp.status_code, data return True, resp.status_code, "ok" except Exception as e: return False, -1, str(e)

健康检查的好处是把失败提前暴露：启动后第一分钟就知道Cookie有没有问题，而不是跑完200个组合之后才发现所有请求都返回验证页。配合异常告警，还可以在健康检查失败时发送通知，这样人不用盯着日志。

5. 社区组合爬虫避坑：字段对不上、IP风控与接口改版的五个实战坑

爬虫写完之后，真正耗时间的是和网站风控、数据结构变化的拉锯。下面这些坑几乎每个做过这类爬虫的人都会遇到。

5.1 组合ID找错：页面上的是展示ID，接口里用的又是另一个ID

现象：你从地址栏复制了组合的ID，比如portfolio/abc123，把它填进接口请求，返回却一直是404或空列表，但页面数据明明正常。

原因：URL里的abc123可能是组合的展示标识，不是接口组合ID。真实ID常常藏在页面源码的全局变量里，或者要调一次元信息接口才返回。这种情况在新版页面里特别多，因为展示ID要短要好记，而内部ID是整数型的。

解决：先在页面源码里搜索portfolioId、portfolio_id、pid这些关键字，找到真实ID。更稳妥的做法是写一个get_real_portfolio_id(session, alias)函数，先请求组合基本信息接口，从返回值里取id，再拼调仓接口。这个解析步骤单独封装，后续接口如果改了映射逻辑，只需改一个函数。

5.2 调仓记录缺字段：价格为"--"、股票名称为空

现象：数据库里出现大量price为None的记录，或者stock_name是空字符串，时间字段格式还是"2023-11-15 10:30:00"这样的字符串。

原因：组合调仓记录并不总是包含成交价，很多模拟盘操作使用的价格字段是空字符串或"--"。停牌股或刚上市标的，名称也可能暂缺。前端页面把它渲染成“价格不变”，但JSON里就是一个空值。

解决：写一个normalize_record函数，把所有脏值统一处理。价格空就填None，名称空就填“未知”，时间字符串转成Unix时间戳。这个函数必须是所有入库记录的唯一入口，千万不要在多个地方各自处理，否则漏掉一个空值就会让统计结果翻车。

5.3 请求频率稍高就触发验证码或滑块

现象：脚本正常运行了半小时，忽然某次请求返回的不是JSON，而是一大段HTML，里面出现类似“验证”“安全”的关键字，状态码依然是200。

原因：网站的风控统计在这段时间内触到了阈值。触发条件不一定是每秒请求数，也可能是单IP或单会话的请求总量。当你开始翻完50个组合后，阈值就到了。

解决：立刻停止脚本，把并发数降到5以下，每页之间等待至少2秒，等待风控状态自然恢复。不要继续硬闯，验证码一旦开始出现，账号或IP可能已经进了更高强度的观察名单，这时候越挣扎越危险。我给异步版默认并发设10，就是希望大多数人从这个安全值开始，而不是从50开始试。

5.4 调仓价格与K线价格对不上

现象：某条调仓记录里价格是15.20，但同一天该股票的行情收盘价是14.85，两个数对不上，一致性校验报错。

原因：接口里的价格不一定是当天实际成交价。它可能是复权后的价格、组合的成本价、或者调仓发生时点的市价快照。不同社区的命名还可能叫price、cost_price、trade_price，含义并不相同。

解决：先确认字段口径。如果只用它做展示，可以直接入库；如果要做收益相关的计算，就要在入库前先决定用什么价格源对齐。我一般会把接口原始价格字段名保留在原字段上，另建一列自己算好的参考价。这样事后发现口径错了，还能追溯原始数据，等于给自己留了后悔药。

5.5 接口改版，字段名和路径全变了

现象：某天代码没改，日志里忽然全是KeyError: 'items'，或者返回里的字段名从stock_code变成了symbol。

原因：网站接口升级了，常见是加了版本号路径，比如/v1/变成/v2/，或者把返回值里的items改成了records。这个变动不会通知外部，只有第二次请求时才会发现。

解决：把解析逻辑隔离在adapter层，业务代码不要直接调raw["data"]["items"]。同时，每次启动时先做一次结构预检，检查关键字段是否存在，不存在就发告警并停止。这样做的好处是，即使接口改版，最多损失一天的数据，不会把解析错误当成“没有调仓”写进库里，留下错误数据后患。

6. 把两个版本变成一套能长期跑的组合调仓监控

6.1 双版本兜底：同步版负责日常，异步版负责批量

两个版本不是竞争关系。我实际的使用方式是把同步版挂在定时任务里，每天固定三次，保证数据按时入库；当需要临时回补数据、或一次拉取大量组合时，再把异步版跑一遍，跑完即停。这样常态请求量低，批量任务又不会影响日常任务。

6.2 定时调度与异常告警

用系统计划任务就能完成调度。

# 周一至周五 09:30 / 13:30 / 15:30 各跑一次同步版 30 9,13,15 * * 1-5 /usr/bin/python3 /home/dev/portfolio_sync.py >> /var/log/portfolio.log 2>&1

日志里最需要记录的是每次抓取成功条数和耗时。我习惯在脚本末尾无论如何都打印一行SUCCESS 条数=xx 耗时=xx或FAILED 原因=xx，这样看日志时只用搜这个标记，就能把一整天的运行情况拉出来。

6.3 合规边界与数据使用习惯

必须承认，任何一个带登录态的数据采集都有合规边界。我的建议是：只抓公开可见的、不需要额外权限就能看到的数据；控制请求频率，不要影响网站正常运行；抓下来的数据只用于个人学习、复盘和分析，不打包传播。我自己跑这类脚本时也翻过车，最开始为了抢实时调仓消息，把异步并发调到30，结果跑了十几分钟就触发校验，账号被限制登录，后来才明白低频持久比一次拉满更重要。希望这个方案能帮你在组合调仓数据采集上少走弯路，也希望你把抓下来的数据用在研究和复盘上，这才是这套工具值得投入的地方。

本文还有配套的精品资源，点击获取