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

pip install pywencai 

QMT_WENCAI├── stock_selector.py    # 选股逻辑├── quant_trader.py      # 交易执行└── config.py           # 配置文件（可选） 

# 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)

# 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()