告别手动选股!用这个Python神器,让同花顺问财自动把数据喂到你嘴边

pip install pywencai

document.cookie

import pywencai# 把刚才复制的Cookie贴在这里MY_COOKIE = 'your_cookie_string_here'# 就这一行！df = pywencai.get(    query='macd金叉，大单流入',  # 你想搜的任何条件    cookie=MY_COOKIE)# 看看结果print(df.head())print(f"共找到 {len(df)} 只股票")

df = pywencai.get(    query='商业航天概念，营收增长大于50%',    cookie=MY_COOKIE,    loop=True,        # 开启自动翻页，抓取全部数据    perpage=100,      # 每页最多100条    sleep=1           # 每页间隔1秒，做个体贴的爬虫)print(df.head())print(f"共找到 {len(df)} 只股票")

df = pywencai.get(    query='kdj金叉 主力净流入>5000万',    cookie=MY_COOKIE,    retry=5   # 失败后重试5次)

df = pywencai.get(    query='创新药概念，研发费用占比>10%',    cookie=MY_COOKIE,    sort_key='研发费用占比',    sort_order='desc'   # 降序)

import osimport timeimport jsonimport pywencaiimport scheduleimport akshare as akimport pandas as pdimport chinese_calendar as ccfrom datetime import datetime, timedeltadef read_json_file(file_path: str) -> dict:    """        读取json文件        Args:            file_path: 文件路径        Returns:            json_object: json对象        """    try:        with open(file_path, "r", encoding="utf-8") as f:            json_object = json.load(f)    except Exception as e:        json_object = {}    return json_objectdef write_json_file(data: dict, file_path: str):    """    写入json文件    Args:        data: 需要写入的数据        file_path: 文件路径    """    os.makedirs(os.path.dirname(file_path), exist_ok=True)    with open(file_path, "w", encoding="utf-8") as f:        json.dump(data, f, ensure_ascii=False, indent=2)def get_previous_day():    today_date = datetime.now().date()    previous_day = today_date - timedelta(days=1)    while previous_day.weekday() >= 5 or cc.is_holiday(previous_day):        previous_day -= timedelta(days=1)    return previous_daydef get_pre_upstop_stock_dict():    try:        stock_zt_pool_previous_em_df = ak.stock_zt_pool_previous_em(            date=datetime.now().strftime("%Y%m%d"))        pre_upstop_stock_list = [            f"{stockcode}.SH" if stockcode.startswith("6") else            f"{stockcode}.SZ" if stockcode.startswith(                ("0", "3")) else f"{stockcode}.BJ"            for stockcode in stock_zt_pool_previous_em_df["代码"]        ]        # 过滤.BJ股票        pre_upstop_stock_list = [            stock for stock in pre_upstop_stock_list            if not stock.endswith(".BJ")        ]        return {            "股票列表": pre_upstop_stock_list,            "统计日期": datetime.now().strftime("%Y%m%d%H%M%S")        }    except Exception as e:        print(f"获取昨日涨停股票列表失败: {e}")        return {"股票列表": [], "统计日期": datetime.now().strftime("%Y%m%d%H%M%S")}def save_pre_upstop_stock_data(save_path: str):    pre_upstop_stock_dict = get_pre_upstop_stock_dict()    write_json_file(data=pre_upstop_stock_dict, file_path=save_path)def get_iwencai_data_dict(cookie: str, question: str) -> dict:    try:        res = pywencai.get(query=question,                           loop=True,                           pro=True,                           cookie=cookie,                           no_detail=True)        return {            "股票列表": res["股票代码"].to_list(),            "统计日期": datetime.now().strftime("%Y%m%d%H%M%S")        }    except Exception as e:        print(f"获取i问财数据失败: {e}")        return {"股票列表": [], "统计日期": datetime.now().strftime("%Y%m%d%H%M%S")}def save_iwencai_selfstock_data(cookie: str, question: str, save_path: str):    iwencai_selfstock_data = get_iwencai_data_dict(cookie=cookie,                                                   question=question)    write_json_file(data=iwencai_selfstock_data, file_path=save_path)if __name__ == "__main__":    strategy_name = "盘中实时选股"    # 数据文件路径    folder_path = os.path.join("D:", "交易策略文件", strategy_name)    # 确保文件夹存在    os.makedirs(folder_path, exist_ok=True)    # i问财cookie    iwencai_cookie = "your_cookie "    # 保存昨日涨停数据    save_pre_upstop_stock_data(os.path.join(folder_path, "昨日涨停数据.json"))    print("昨日涨停数据已保存")    # 定义i问财选股结果数据保存路径    iwencai_selfstock_data_path = os.path.join(folder_path, "i问财选股结果.json")    # 创建一个不带参数的包装函数，用于定时任务    def save_iwencai_selfstock_data_job():        question = "竞价异动类型包含竞价抢筹，集合竞价评级包含看多，dde大单净额大于0，今日涨幅大于3%，沪深A股，非ST，非次新股"        save_iwencai_selfstock_data(cookie=iwencai_cookie,                                    question=question,                                    save_path=iwencai_selfstock_data_path)        print(f"i问财选股结果已保存: {datetime.now().strftime('%H:%M:%S')}")    # 立即执行一次i问财选股结果保存    save_iwencai_selfstock_data_job()    # 设置定时任务 - 每30秒执行一次    interval_seconds = 30  # 可以修改这个值来调整执行间隔    schedule.every(interval_seconds).seconds.do(        save_iwencai_selfstock_data_job)    print("定时任务已启动，每分钟保存i问财选股结果")    while True:        schedule.run_pending()        # 显示下一次执行时间        next_run = schedule.next_run()        if next_run:            remaining = next_run - datetime.now()            print(f"\r下一次i问财选股结果获取将在 {int(remaining.total_seconds())} 秒后",                  end="")        time.sleep(1)