环境准备：工欲善其事，必先利其器

说实话，做量化系统这么多年，我见过太多人一上来就写代码，结果环境配了三天还没跑通。嗯，咱们别走那条弯路。这一章，我把环境准备这件事掰开揉碎了讲清楚。

我个人习惯是，每做一个新项目，都从零搭建一套干净的环境。为什么？因为依赖冲突这玩意儿，真的能让你怀疑人生。我曾经在一个项目里，因为numpy版本不对，硬是debug了两天。

Python 3.10+：为什么选它？

Python 3.10 引入了一个很重要的特性——结构模式匹配（就是那个 match-case）。你想想看，处理Tick数据时，经常要根据不同的消息类型走不同的逻辑，用 if-elif-else 写出来又臭又长。match-case 让代码清晰很多。

另外，3.10 的性能优化也不错。尤其是类型提示的增强，对咱们做量化系统的人来说，代码可读性就是生命。

下载地址我就不贴了，去 python.org 找就行。安装时记得勾选「Add Python to PATH」，这个坑我踩过——没勾选的话，命令行里打 python 会提示找不到命令。

虚拟环境管理：venv vs conda

说白了，虚拟环境就是给你的每个项目一个独立的小房间。A项目用 pandas 1.5，B项目用 pandas 2.0，互不干扰。没有虚拟环境？那你就等着「依赖地狱」吧。

venv：轻量级选手

Python 自带的，不用额外装。我个人在 Linux 和 macOS 上用得最多。

# 创建虚拟环境
python -m venv tick_env

# 激活（Windows）
tick_env\Scripts\activate

# 激活（macOS/Linux）
source tick_env/bin/activate

# 退出
deactivate

注意：venv 创建的虚拟环境，默认不会继承系统级的包。这是好事，干净。

conda：重型武器

如果你做数据分析或者机器学习，conda 会更顺手。它能管理 Python 版本，还能装一些非 Python 的依赖（比如 C 库）。

# 创建环境并指定 Python 版本
conda create -n tick_env python=3.10

# 激活
conda activate tick_env

# 退出
conda deactivate

核心依赖库介绍

咱们这个 Tick 数据采集系统，说白了就靠这几个库撑起来。我一个个说。

库名	版本建议	作用
pandas	≥ 2.0	数据处理、时间序列分析
numpy	≥ 1.24	数值计算、数组操作
aiohttp	≥ 3.8	异步 HTTP 请求（对接行情接口）
motor	≥ 3.1	异步 MongoDB 驱动（存 Tick 数据）
redis	≥ 4.5	缓存、消息队列（实时数据中转）

pandas：数据处理的瑞士军刀

做量化的，没人离得开 pandas。Tick 数据进来后，你要做清洗、对齐、重采样，pandas 一套带走。

我记得有一次，行情接口返回的时间戳是字符串格式，还带时区信息。用 pandas 的 to_datetime() 一行搞定。要是自己写解析逻辑，至少 20 行。

import pandas as pd

# 读取 Tick 数据（示例）
df = pd.read_csv('tick_data.csv', parse_dates=['timestamp'])
df.set_index('timestamp', inplace=True)

# 重采样到 1 分钟 K 线
minute_data = df['price'].resample('1min').ohlc()

numpy：性能基石

pandas 底层就是 numpy。你处理百万级的 Tick 数据，循环肯定不行，得用向量化操作。numpy 就是干这个的。

import numpy as np

# 计算 Tick 数据的收益率（向量化）
prices = np.array([100.0, 100.5, 101.2, 100.8])
returns = np.diff(prices) / prices[:-1]  # 一行搞定

aiohttp：异步 HTTP 请求

行情接口通常都是 WebSocket 或者 HTTP 轮询。如果是 HTTP，用 requests 库会阻塞，一个请求没回来，后面的都得等着。aiohttp 是异步的，可以同时发几十个请求，效率天差地别。

import aiohttp
import asyncio

async def fetch_tick(session, url):
    async with session.get(url) as resp:
        return await resp.json()

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_tick(session, f'http://api.quote.com/tick?code={c}') 
                 for c in ['000001', '000002', '000003']]
        results = await asyncio.gather(*tasks)
        print(results)

asyncio.run(main())

motor：异步 MongoDB 驱动

Tick 数据量很大，一天几百万条很正常。MongoDB 这种文档数据库很适合存 Tick——每条数据就是一个文档，字段灵活。motor 是异步版的 pymongo，配合 aiohttp 用，整个采集链路都是异步的，不会阻塞。

import motor.motor_asyncio

client = motor.motor_asyncio.AsyncIOMotorClient('mongodb://localhost:27017')
db = client.tick_db
collection = db.tick_data

# 异步插入一条 Tick
async def insert_tick(data):
    result = await collection.insert_one(data)
    return result.inserted_id

redis：实时数据中转站

采集到的 Tick 数据，有时候需要立刻推送给策略模块或者前端展示。直接写数据库太慢，redis 作为内存数据库，读写都是微秒级。我通常用 redis 的 List 或者 Pub/Sub 来做实时数据管道。

import redis.asyncio as redis

async def push_tick(redis_client, key, data):
    await redis_client.lpush(key, data)  # 推入列表头部
    await redis_client.ltrim(key, 0, 999)  # 只保留最近 1000 条

本章知识体系

下面这张图，把咱们这一章的核心逻辑串起来了。你看一眼，心里就有数了。

这张图你看懂了吗？从上到下，先装 Python，再建虚拟环境，最后装依赖库。顺序别搞反了。我见过有人先装库再建虚拟环境，结果库装到了系统级，虚拟环境里啥也没有。

---