企业微信

环境配置与接口对接

在Windows系统中安装easytrader前需完成Python3.7+环境部署，通过pip install easytrader命令获取最新版本。同花顺i问财开放平台需注册开发者账号并创建应用，获取APP_KEY和APP_SECRET参数。配置过程中需修改easytrader源码中的券商配置文件，将同花顺的API地址、端口信息及认证参数写入配置项。

交易环境测试可通过以下代码验证：

如何用Python框架实现同花顺自动化交易？


import easytrader

user = easytrader.use('ths')

user.prepare('account.json')

其中account.json需包含同花顺账户信息及交易单元代码。测试过程中应监控日志文件，确认是否出现认证超时或接口调用失败等异常。

核心交易功能代码实现

买卖委托指令可通过封装后的函数实现：


def send_order(code, price, amount, entrust_bs):

    """code:证券代码, price:委托价格, amount:委托数量, entrust_bs:买卖方向"""

    result = user.trades(

        stock_code=code,

        price=price,

        amount=amount,

        entrust_bs=entrust_bs

    )

    return result['entrust_no'] if result['status'] == 'success' else None

撤单功能需记录委托编号，通过循环扫描未成交订单：


def cancel_pending_orders():

    orders = user.query_history_orders(status='pending')

    for order in orders:

        user.cancel_order(entrust_no=order['entrust_no'])

实时行情监控与策略执行

使用同花顺实时行情接口需建立WebSocket连接，订阅指定证券代码的行情数据：


import websockets

async def receive_quote():

    async with websockets.connect('wss://marketdata.10jqka.com:8080') as websocket:

        await websocket.send(json.dumps({'codes':['sh600000','sz000001'],'token':user.token}))

        while True:

            data = await websocket.recv()

            process_quote_data(data)

行情处理函数需实时计算技术指标，当MACD金叉且成交量突破日均200%时触发买入信号：


def process_quote_data(data):

    quotes = json.loads(data)

    for quote in quotes:

        if quote['macd'] > 0 and quote['volume'] > quote['avg_volume']*2:

            entrust_no = send_order(quote['code'], quote['price'], 1000, 'B')

风控模块与异常处理

构建三层风控体系：资金校验层在下单前检查可用余额，通过user.balance()接口获取实时资金数据；委托校验层验证价格合理性，防止乌龙指；系统校验层监控网络延迟，当API响应超时超过3次时自动暂停交易。

异常处理机制需包含：

  1. 重试策略：对网络超时采用指数退避算法重试3次

  2. 熔断机制：单日亏损超过5%时暂停交易

  3. 日志记录：使用logging模块记录每笔委托的完整生命周期


import logging

logging.basicConfig(filename='trading.log', level=logging.INFO)

try:

    entrust_no = send_order('sh600000', 15.5, 1000, 'B')

except Exception as e:

    logging.error(f"Order failed: {str(e)}", exc_info=True)

策略优化与回测验证

通过历史数据回测验证策略有效性，使用pandas处理同花顺导出的历史行情：


import pandas as pd

df = pd.read_csv('history_data.csv')

df['signal'] = (df['close'] > df['ma20']) & (df['volume'] > df['volume'].rolling(20).mean()*1.5)

绩效评估指标包括夏普比率、最大回撤、胜率等：


def calculate_metrics(returns):

    sharpe_ratio = np.mean(returns)/np.std(returns)*np.sqrt(252)

    max_drawdown = np.max(np.maximum.accumulate(returns) - returns)

    win_rate = len([r for r in returns if r>0])/len(returns)

    return {'sharpe':sharpe_ratio, 'max_drawdown':max_drawdown, 'win_rate':win_rate}

实盘运行前需进行模拟盘测试，确保API调用频率符合交易所规定，同花顺接口限制为每秒50次请求，可通过令牌桶算法控制请求速率：


import time

class RateLimiter:

    def __init__(self, max_tokens, refill_rate):

        self.tokens = max_tokens

        self.max_tokens = max_tokens

        self.refill_rate = refill_rate

        self.last_time = time.time()

    def consume(self, tokens=1):

        now = time.time()

        self.tokens = min(self.max_tokens, self.tokens + (now - self.last_time)*self.refill_rate)

        if self.tokens >= tokens:

            self.tokens -= tokens

            self.last_time = now

            return True

        return False