Py学习  »  Python

如何使用 Python 获取同花顺问财数据?全网最简单方法!

AirPython • 1 月前 • 528 次点击  
一、环境准备
1.1 软件安装与配置
同花顺i问财:
安装 pywencai 
https://gitcode.com/gh_mirrors/py/pywencai?utm_source=highlight_word_gitcode&word=pywencai&isLogin=1
pip install pywencai 
图片

视频教程:

https://www.bilibili.com/video/BV1NP411C7UU/?vd_source=11a82a19c33a01b5927c511379211088

迅投QMT:
通过合作券商获取QMT安装包
安装并完成基础配置
申请模拟盘或者实盘交易权限(需满足券商要求)
安装vscode,创建以下项目:
图片
QMT_WENCAI├── stock_selector.py    # 选股逻辑├── quant_trader.py      # 交易执行└── config.py           # 配置文件(可选) 
二、数据获取与处理
2.1 同花顺数据接口
历史数据获取:使用vscode ,创建新项目,如QMT_WENCAI,然后创建选股模块,供qmt使用
图片
比如以下:
# stock_selector.pyimport pywencaidef select_stocks():    """    选股主函数,返回符合以下条件的股票:    1. 最近5日有过涨停    2. 最近5日没有跌停    3. 今日成交量 > 5日平均成交量    4. 今日集合竞价涨幅在2%-3%之间    5. 排除北交所、科创板、创业板、ST股票    返回:        list: 符合条件的股票代码列表(如['000001.SZ']),如果没有符合条件的股票则返回空列表    """    try:        # 构造查询语句        query = (            "最近5日有过涨停,"            "最近5日没有跌停,"            "今日成交量>5日平均成交量,"            "2<今日集合竞价涨幅<3,"            "非北交所非科创板非创业板非ST"        )        # 调用 pywencai 获取数据        df = pywencai.get(query=query, sort_key='成交金额', sort_order='desc')        if df is None or df.empty:            print("[StockSelector] 未找到符合条件的股票。")            return []        # 返回股票代码列表        if '股票代码' in df.columns:            return df['股票代码'].tolist()        elif '代码' in df.columns:            return df['代码'].tolist()        else:            print("[StockSelector] 错误:数据框中找不到股票代码列。")            return []    except Exception as e:        print(f"[StockSelector] 查询时发生错误: {e}")        return []# 示例调用if __name__ == "__main__":    selected_stocks = select_stocks()    print("符合条件的股票:", selected_stocks)
返回结果如下:
图片
接着,使用qmt进行下单,打开券商提供的qmt,输入账号密码,勾选下面独立交易,就可以使用miniqmt了。
图片

三、策略开发和下单
基于同花顺的选股结果,我们就可以继续开发策略了,quant_trader.py代码如下,其中 ACCOUNT_ID替换成自己的账号即可:
# quant_trader.pyfrom xtquant.xttrader import XtQuantTrader, XtQuantTraderCallbackfrom xtquant.xttype import StockAccountfrom xtquant import xtconstantimport timefrom datetime import datetimefrom xtquant import xtdatafrom stock_selector import select_stocks  # 直接导入选股函数class MyXtQuantTraderCallback(XtQuantTraderCallback):    def __init__(self):        super().__init__()        self.last_order_id = None    def on_disconnected(self):        print(f"[{datetime.now()}] 连接断开")    def on_stock_order(self, order):        self.last_order_id = order.order_id        print(f"[{datetime.now()}] 委托回报: {order.stock_code} 状态:{order.order_status}")    def on_stock_trade(self, trade):        print(f"[{datetime.now()}] 成交回报: {trade.stock_code} 价格:{trade.traded_price} 数量:{trade.traded_volume}")    def on_order_error(self, order_error):        print(f"[{datetime.now()}] 委托失败: {order_error.error_msg}")    def on_cancel_error(self, cancel_error):        print(f"[{datetime.now()}] 撤单失败: {cancel_error.error_msg}")class QuantTrader:    def __init__(self, qmt_path, session_id, account_id):        self.xt_trader = XtQuantTrader(qmt_path, session_id)        self.acc = StockAccount(account_id)        self.callback = MyXtQuantTraderCallback()        self.connected = False    def connect(self):        try:            self.xt_trader.register_callback(self.callback)            self.xt_trader.start()            connect_result = self.xt_trader.connect()            if connect_result != 0:                raise ConnectionError(f"交易服务器连接失败,错误码: {connect_result}")            subscribe_result = self.xt_trader.subscribe(self.acc)            if subscribe_result != 0:                raise ConnectionError(f"账号订阅失败,错误码: {subscribe_result}")            self.connected = True            print(f"[{datetime.now()}


    
] 交易服务器连接成功")            return True        except Exception as e:            print(f"[{datetime.now()}] 连接失败: {e}")            return False    def get_realtime_price(self, stock_code):        """获取股票实时价格"""        try:            quote = xtdata.get_full_tick([stock_code])            if quote and stock_code in quote:                return quote[stock_code]['lastPrice']        except Exception as e:            print(f"[{datetime.now()}] 获取实时价格失败: {e}")        return None    def place_order(self, stock_code, price, amount, order_direction):        """下单函数"""        try:            order_id = self.xt_trader.order_stock(                self.acc,                 stock_code,                 order_direction,                 amount,                 xtconstant.FIX_PRICE,                 price,                 '涨停策略'                '系统自动下单'            )            print(f"[{datetime.now()}] 下单成功: {stock_code} 方向:{order_direction} 价格:{price} 数量:{amount}")            return order_id        except Exception as e:            print(f"[{datetime.now()}] 下单失败: {e}")            return None    def execute_buy(self, stock_code, amount=100):        """执行买入操作"""        price = self.get_realtime_price(stock_code)        if price is None:            print(f"[{datetime.now()}] 无法获取实时价格,取消买入")            return False        return self.place_order(stock_code, price, amount, xtconstant.STOCK_BUY) is not None    def run_strategy(self):        """执行策略"""        try:            if not self.connect():                return            print(f"[{datetime.now()}] 开始执行选股策略...")            selected_stocks = select_stocks()  # 直接调用选股函数            if not selected_stocks:                print(f"[{datetime.now()}] 今日未选出符合条件的股票")                return            print(f"[{datetime.now()}] 选出的股票: {selected_stocks}")            # 对每只选出的股票执行买入            for stock_code in selected_stocks:                if self.execute_buy(stock_code):                    print(f"[{datetime.now()}


    
] 成功买入: {stock_code}")                else:                    print(f"[{datetime.now()}] 买入失败: {stock_code}")        except KeyboardInterrupt:            print(f"[{datetime.now()}] 用户中断程序")        except Exception as e:            print(f"[{datetime.now()}] 策略执行出错: {e}")        finally:            print(f"[{datetime.now()}] 程序结束")if __name__ == "__main__":    # 配置参数(根据实际情况修改)    QMT_PATH = 'C:\\国金QMT交易端模拟\\userdata_mini'    SESSION_ID = 123456    ACCOUNT_ID = '*******'    print(f"[{datetime.now()}] 启动量化交易程序")    trader = QuantTrader(QMT_PATH, SESSION_ID, ACCOUNT_ID)    trader.run_strategy()
执行结果如下:
图片
切换到qmt,可以看到我们刚才选出并下单的股票已经成功在qmt下单:
图片
通过同花顺和QMT的组合,我们可以构建从数据获取、策略研发到实盘交易的完整量化交易体系。本文介绍的框架和方法可以作为起点,实际应用中需要根据个人风险偏好和市场特点进行调整。量化交易的核心在于持续优化和严格的风险控制,希望本文能为您的量化交易之旅提供有价值的参考。
————————————————
版权声明:本文为CSDN博主「NI_LINUX」的原创文章,遵循CC 4.0 BY-SA版权协议,转载请附上原文出处链接及本声明。
文章链接:

https://blog.csdn.net/NI_LINUX/article/details/147233601

Python社区是高质量的Python/Django开发社区
本文地址:http://www.python88.com/topic/183539
 
528 次点击