RateLimitingIterator 是一个轻量级秒级限流迭代器，每秒最多处理 N 条数据，通过 time.time() 滑动控制间隔、自动 sleep 补时，支持任意可迭代对象，适用于 Tushare 等配额受限 API 的匀速调用。
直接用自定义的 ratelimitingiterator 控制数据遍历节奏，核心是把“取数据”和“等时间”封装进迭代逻辑里，让每次 next() 都自动守时。它不依赖外部中间件或全局状态，适合本地脚本、批量拉取、爬虫、对接有配额限制的 api（如 tushare、instagram、大模型接口）等场景。
设计一个轻量级秒级限流迭代器
目标：每秒最多处理 N 条数据，不堆积、不超发、不阻塞主线程逻辑。下面是一个基于时间戳滑动判断的简洁实现：
- 使用
time.time()记录上一次处理时间 - 每次迭代前计算距上次的间隔，若不足 1/N 秒，则 sleep 补足
- 支持传入任意可迭代对象（列表、生成器、数据库游标等）
- 线程安全非必需，但加锁可适配多线程消费场景
示例代码：
import time
from typing import Iterator, Any
<p>class RateLimitingIterator:
def <strong>init</strong>(self, iterable: Iterator[Any], per_second: int = 1):
self.iterable = iter(iterable)
self.interval = 1.0 / max(1, per_second) # 每条最小间隔（秒）
self.last_called = 0.0</p><pre class="brush:php;toolbar:false;"><pre class="brush:php;toolbar:false;">def __iter__(self):
return self
def __next__(self):
now = time.time()
elapsed = now - self.last_called
if elapsed < self.interval:
time.sleep(self.interval - elapsed)
self.last_called = time.time()
return next(self.iterable)</code>实战：控制 Tushare 财务数据分批拉取频率
Tushare 免费用户常受限于「每分钟 60 次」或「单次最多 60 条」，用该迭代器可轻松拆解大批量日期范围，并匀速调用：
- 先构造所有待查日期（如过去 5 年每季度首日）
- 将日期列表传入
RateLimitingIterator，设 <code>per_second=1→ 实际约 60 次/分钟 - 循环中每次只 fetch 一个日期的数据，天然规避并发超限
片段示意：
Evez Github Manager
下载
在 OpenClaw 中管理 GitHub 仓库、Pull Request、Issue 与工作流。适用于审查 PR、管理 Issue、自动化 GitHub Actions、同步仓库等场景。
from datetime import datetime, timedelta
import tushare as ts
<h1>构造季度日期列表（2020Q1 ~ 2025Q1）</h1><p>dates = []
for year in range(2020, 2026):
for quarter in [1, 4, 7, 10]:
dt = datetime(year, quarter, 1)
dates.append(dt.strftime('%Y%m%d'))</p><h1>匀速遍历调用</h1><p>rl_iter = RateLimitingIterator(dates, per_second=1)
pro = ts.pro_api()</p><p>for trade_date in rl_iter:
try:
df = pro.fina_indicator(ts_code='000001.SZ', period=trade_date[:6])
print(f"Fetched {trade_date}: {len(df)} rows")
except Exception as e:
print(f"Failed on {trade_date}: {e}")搭配异步协程提升吞吐（进阶用法）
若需更高效率又不突破频控，可将限流逻辑下沉到异步任务内部 —— 即每个协程自己守时，再用 asyncio.gather 并发多个低频通道：
- 例如启动 3 个协程，每个按 0.5 QPS（即每 2 秒 1 次），整体稳定在 1.5 QPS
- 比单线程串行快 3 倍，仍严格满足服务端限流要求
- 避免因某次请求慢拖累全局节奏
关键点：限流器必须在协程内实例化，不可跨协程共享计时状态。
注意边界与容错细节
真实场景中容易忽略但影响稳定性的问题：
-
首次启动偏差：初始化时
last_called=0会导致第一条立即执行，建议设为time.time() - interval -
系统时间跳变：NTP 校准可能使
time.time()回跳，可改用time.monotonic()更鲁棒 -
异常中断后恢复：若某次请求失败并跳过
next()，下次仍会补时；如需跳过限流，应在捕获异常后手动调用next()并丢弃 -
精度取舍：
time.sleep()在 Linux 下通常误差 ±10ms，对秒级控制足够；毫秒级需用asyncio.sleep()或专用调度库
