2、开发环境搭建：Python版本选择、虚拟环境配置、核心依赖库安装

说实话，搭建开发环境这件事，很多新手觉得不值一提。

但我见过太多人，花了两天时间调一个bug，最后发现是Python版本不对。或者依赖库冲突，搞得整个项目跑不起来。嗯，这些坑我都踩过。

今天我们就来把环境搭得干干净净，后面写代码才能顺风顺水。

2.1 Python版本选择：别追新，要追稳

我个人习惯，做量化交易系统，Python版本选3.9或3.10。

为什么不是3.12？你想想看，asyncio在3.9就已经很成熟了，pika、redis-py这些库对3.9的支持也最稳定。我去年有个项目，团队里有人非要用3.11，结果某个异步库的协程调度出了兼容性问题，排查了两天才发现是版本锅。

所以我的建议是：Python 3.9.13 或 Python 3.10.11，这两个版本经过大量生产环境验证，最靠谱。

2.2 虚拟环境配置：隔离是美德

我曾经在一个服务器上同时维护三个量化项目，每个项目依赖的库版本都不一样。如果没有虚拟环境，那简直就是噩梦。

Python自带的 venv 就够用了。我一般这样操作：

# 创建虚拟环境
python3.9 -m venv trading_env

# 激活（Linux/Mac）
source trading_env/bin/activate

# 激活（Windows）
trading_env\Scripts\activate

# 退出
deactivate

激活后，终端前面会出现 (trading_env) 字样，说明你已经在虚拟环境里了。这时候安装的所有库，都只在这个环境里生效，不会污染系统全局。

2.3 核心依赖库安装

好，虚拟环境准备好了，接下来装核心库。这四个库是我们事件驱动系统的骨架：

库名	版本建议	作用
asyncio	Python 3.9+ 内置	事件循环、协程调度
pika	1.3.2	RabbitMQ 客户端，消息队列
redis	5.0.0	Redis 客户端，缓存/状态存储
sqlalchemy	2.0.23	ORM，数据库操作

安装命令很简单：

pip install pika==1.3.2 redis==5.0.0 sqlalchemy==2.0.23

asyncio 不需要单独安装，Python 3.9 自带。

2.3.1 asyncio：事件循环的心脏

说白了，asyncio 就是我们的「调度中心」。所有的事件处理、消息收发、定时任务，都靠它来协调。

我刚开始用 asyncio 时，总觉得它和普通的同步代码差不多。后来才发现，协程不是函数，你得用 await 去调用它。这个坑我踩了不止一次。

import asyncio

async def handle_event(event):
    print(f"处理事件: {event}")
    await asyncio.sleep(0.1)  # 模拟IO操作
    return "完成"

async def main():
    result = await handle_event("订单成交")
    print(result)

asyncio.run(main())

2.3.2 pika：连接RabbitMQ的桥梁

pika 是 Python 操作 RabbitMQ 最常用的库。事件驱动系统里，消息队列就是「血管」，pika 就是连接血管的管道。

我记得第一次用 pika 时，被它的 BlockingConnection 和 SelectConnection 搞晕了。简单说：

• BlockingConnection：同步模式，简单但会阻塞
• SelectConnection：异步模式，配合 asyncio 使用

我们做事件驱动系统，当然选异步模式。

import pika
import asyncio

# 异步连接示例（简化版）
async def connect_rabbitmq():
    connection = await pika.connect("amqp://guest:guest@localhost/")
    channel = await connection.channel()
    return channel

2.3.3 redis：缓存与状态存储

Redis 在我们系统里干两件事：

1. 缓存：存行情数据、中间计算结果
2. 状态存储：记录订单状态、账户信息

用 redis-py 这个库，配合 asyncio 的 aioredis（现在合并到 redis-py 里了）。

import redis.asyncio as aioredis

async def get_redis():
    r = await aioredis.Redis(host='localhost', port=6379, db=0)
    await r.set('order:123', 'filled')
    value = await r.get('order:123')
    print(value)  # b'filled'
    await r.close()

2.3.4 sqlalchemy：数据库操作的门面

SQLAlchemy 2.0 版本改进了异步支持，可以直接用 async_session。我们用它来操作 MySQL 或 PostgreSQL，存储交易记录、策略参数等持久化数据。

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "mysql+aiomysql://user:pass@localhost/trading"

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def save_order(order_data):
    async with async_session() as session:
        session.add(order_data)
        await session.commit()

2.4 知识体系总览

下面这张图，帮你理清这四个库在事件驱动系统里的位置：

看到没？asyncio 在中间，像心脏一样泵送事件。pika 负责从外部接收消息，redis 负责快速存取状态，sqlalchemy 负责把重要数据落盘。各司其职，互不干扰。

2.5 验证环境是否搭好

装完库，跑个简单的测试脚本，确认一切正常：

import asyncio
import pika
import redis.asyncio as aioredis
from sqlalchemy import create_engine

async def test_env():
    print("asyncio: OK")
    
    # 测试 pika（需要 RabbitMQ 运行）
    # print("pika: OK")
    
    # 测试 redis
    try:
        r = await aioredis.Redis()
        await r.ping()
        print("redis: OK")
        await r.close()
    except:
        print("redis: 连接失败，请检查 Redis 服务")
    
    # 测试 sqlalchemy
    try:
        engine = create_engine("sqlite:///test.db")
        engine.connect()
        print("sqlalchemy: OK")
    except:
        print("sqlalchemy: 连接失败")

asyncio.run(test_env())

如果输出都是 OK，恭喜你，环境搭好了。

---

公众号:蓝海资料掘金营，微信deep3321