最近有几个同学想实现聚宽选股，pt或者qmt之类的第三方软件交易的目的，前后发了一些代码过去，今日有空，随手整理一份比较完整的发出来，供新手参考，高手可以略过，个人代码功底欠佳，有砖轻拍。 一、思路 实现聚宽选股，第三方交易的核心思路有两种： 第一种：聚宽实时模拟生成预交易股票池（比如蒋老师的小市值10只），通过数据接口给到第三方******，第三方******再编写下单逻辑，即把聚宽模拟里交易的部分交给第三方******完成，仓位统计等在第三方******里完成。 第二种：聚宽策略不变，实时同步聚宽模拟的仓位，第三方******在聚宽策略完成后的一分钟同步聚宽持仓，然后实现调仓。 个人倾向第二种思路。 二、实现方法 第一种：第三方******定时定点读取聚宽策略持仓内容，方法有很多，这里就不描述了，这种办法成本比较低，但是不稳定，并不推荐。 第二种：聚宽通过sqlalchemy模块向中间数据库写入持仓数据（中间数据库可以是云数据库，也可以是通过DDNS映射服务端口出去的本地数据库，我个人交易用云服务器自己搭建，这样比较稳定），找个可以外联数据库的PTrader实时访问数据库，同步持仓数据。 三、核心代码（数据库操作代码） 1.引用模块 ``` from sqlalchemy import create_engine from sqlalchemy.sql import text from sqlalchemy.types import * from contextlib import contextmanager import socket ``` 2.数据库操作类（此代码来自于一溜烟老师，在此感谢老师的帮助） ``` class Store(object): def __init__(self): pass def save(self,df,table_name): try: with get_db_engine() as engine: df.to_sql(table_name, engine, if_exists='append', index=False) except: print("save error") def get_data(self,sql): with get_db_engine() as engine: return pd.read_sql(sql, engine) def exeute(self,sql): with get_db_engine() as engine: with engine.connect() as con: con.execute(sql) ``` 3.数据库写入代码 ``` def write_position_to_sql(context): pos_list_temp = [] pos_list = ['000001.XSHG'] sell_list = list(context.portfolio.positions.keys()) for stock in sell_list: pos_list.append(stock) i = 0 for stock in pos_list: i += 1 pos_list_temp.append( {'stock':stock,'rank_order':i,'date':context.current_dt.strftime("%Y-%m-%d")} ) df_stock2=pd.DataFrame(pos_list_temp) store=Store() store.save(df_stock2,"stocklist") store.exeute("delete from stocklist where date='{date}'".format(date=context.current_dt.strftime("%Y-%m-%d"))) store.save(df_stock2,"stocklist") ``` 这里需要解释一下： 1.sqlalchemy在操作数据库的时候不能写入空键值，所以用了个比较笨的办法，对context.portfolio.positions.keys()生成的list里添加上证指数，防止数据库意外退出。 2.为什么要写两次写入？因为mysql写入的时候不能覆盖写入（ps:我代码能力有限不知道怎么覆写），所以为了防止多次回测写入重复数据，我加了个删除的逻辑，这样无论回测多少次保证数据库里只有一份数据。 4.微信推代码 ``` def send_wechat(msg,flag): token = ' ' # pushplus的token title = flag content = msg template = 'html' topic = '1' url = f"http://www.pushplus.plus/send?token={token}&title={title}&content={content}&template={template}&topic={topic}" #print(url) r = requests.get(url=url) def sell_message(context,stock): current_data = get_current_data() msg = context.current_dt.strftime("%Y-%m-%d") + '日卖出:[%s]%s,成交单价::%.2f'%(stock,current_data[stock].name,current_data[stock].last_price) if len(msg)>20 and g.message: msg = str(msg) send_wechat(msg,'卖出记录') print(msg) def buy_message(context,stock): current_data = get_current_data() msg = context.current_dt.strftime("%Y-%m-%d") + '日买入:[%s]%s,成交单价::%.2f,仓位1/3'%(stock,current_data[stock].name,current_data[stock].last_price) if len(msg)>20 and g.message: msg = str(msg) send_wechat(msg,'买入记录') ``` 我个人比较懒，加个微信推送，如果没有收到消息那就是脚本出问题了，上pt看看日志。 5.pt端代码： 5.1引入模块 ``` import pymysql from sqlalchemy import create_engine from sqlalchemy.sql import text from sqlalchemy.types import * pymysql.install_as_MySQLdb() ``` 5.2读取数据库代码： ``` def get_sql_list(): store=Store() df = store.get_data("select * from stocklist where date='{date}'".format(date=context.current_dt.strftime("%Y-%m-%d"))) pos_list=list(df.sort_values("rank_order")['stock'].values) for i in range(len(pos_list)): pos_list[i] = pos_list[i].replace("XSHE", "SZ") pos_list[i] = pos_list[i].replace("XSHG", "SS") print(pos_list) ``` pt里也要引入上述Store()类代码 至此，聚宽的实时模拟可以通过中间数据库同步到任意可以外联数据库的第三方交易平台，实现更多的交易可能。 ps：回测里放的是全部的聚宽代码，包括数据库连接、写入，交易策略用的是95老师的小市值，实盘也在跑的，最近收益还可以，核心的操作的代码文章里已经全部写明了，发个回测主要是骗骗积分^_^，不太懂的同学可以看看回测的代码就明白了，我实在太穷了，明年实盘模拟的积分都快不够了，所以分低的同学参考文章里的代码即可，有分的大佬请不吝克隆，感谢感谢。 10.09补充： 关于云服务器，买能买到的最便宜的就行，我个人用的1核1G，5M流量的，足够用了。就是个中转而已，不需要性能多高。云服务器用的centos系统，其实只要是个类linux就行。为了省事儿安装的宝塔面板，配置如图：