[Python技术] 通过同花顺判断股票是否是近期解禁股技术方案

akshare.stock_restricted_release_detail_em(start_date="20250603", end_date="20250703")

from io import StringIO
import pandas as pd
import requests
from bs4 import BeautifulSoup
import py_mini_racer
from akshare.utils.tqdm import get_tqdm
from akshare.datasets import get_ths_js
# 设置pandas显示选项
pd.set_option('display.unicode.ambiguous_as_wide', True)
pd.set_option('display.unicode.east_asian_width', True)
pd.set_option('display.max_rows', None)
pd.set_option('display.max_columns', None)
pd.set_option('display.expand_frame_repr', False)
pd.set_option('display.max_colwidth', 100)
def _get_file_content_ths(file: str = "ths.js") -> str:
    """获取JS文件内容"""
    setting_file_path = get_ths_js(file)
    with open(setting_file_path, encoding="utf-8") as f:
        return f.read()
def get_recent_unlock_stocks():
    """
    获取最近解禁的股票列表（带6位股票代码）
    返回包含股票代码和解禁信息的DataFrame
    """
    js_code = py_mini_racer.MiniRacer()
    js_content = _get_file_content_ths("ths.js")
    js_code.eval(js_content)
    v_code = js_code.call("v")
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.85 Safari/537.36",
        "hexin-v": v_code,
        "X-Requested-With": "XMLHttpRequest"
    }
    # 获取总页数
    base_url = "https://data.10jqka.com.cn/market/xsjj/field/enddate/order/desc/ajax/1/page/{}/"
    response = requests.get(base_url.format(1), headers=headers)
    soup = BeautifulSoup(response.text, 'lxml')
    page_info = soup.find("span", class_="page_info")
    total_pages = int(page_info.text.split("/")[1]) if page_info else 1
    # 获取所有解禁股票数据
    all_unlock_stocks = pd.DataFrame()
    tqdm = get_tqdm()
    for page in tqdm(range(1, total_pages + 1), desc="获取解禁数据"):
        # 更新hexin-v（重要反爬机制）
        js_code = py_mini_racer.MiniRacer()
        js_code.eval(js_content)
        v_code = js_code.call("v")
        headers["hexin-v"] = v_code
        # 获取当前页数据
        page_url = base_url.format(page)
        response = requests.get(page_url, headers=headers)
        page_df = pd.read_html(StringIO(response.text))[0]
        if isinstance(page_df.columns, pd.MultiIndex):
            page_df.columns = page_df.columns.get_level_values(0)  # 取第一级列名
            # 2. 处理列名重复
        if page_df.columns.duplicated().any():
            page_df = page_df.loc[:, ~page_df.columns.duplicated()]  # 保留唯一列名
        #print(page_df)
        # 处理股票代码为6位格式
        if "股票代码" in page_df.columns:
            # 正确用法：先取单列Series，再使用.str
            stock_series = page_df["股票代码"].astype(str)  # 转换为字符串类型Series
            page_df["6位代码"] = stock_series.str.zfill(6)  # 在Series上使用.str
        all_unlock_stocks = pd.concat([all_unlock_stocks, page_df], ignore_index=True)
    return all_unlock_stocks
def is_recent_unlock(code, unlock_df=None):
    """
    判断股票是否在最近解禁列表中
    :param code: 股票代码（支持多种格式）
    :param unlock_df: 可选的解禁数据DataFrame（避免重复获取）
    :return: (是否解禁, 解禁信息)
    """
    # 统一处理为6位字符串格式
    code_6 = str(code).zfill(6)
    # 获取解禁数据（如果未提供）
    if unlock_df is None:
        unlock_df = get_recent_unlock_stocks()
    # 检查股票是否在解禁列表中
    if unlock_df is not None and "6位代码" in unlock_df.columns:
        match = unlock_df[unlock_df["6位代码"] == code_6]
        if not match.empty:
            return True, match.iloc[0].to_dict()
    return False, None
# 使用示例
if __name__ == "__main__":
    # 获取所有解禁股票数据
    unlock_df = get_recent_unlock_stocks()
    print(unlock_df)
    # 测试股票代码（注意：1309会被处理为001309）
    test_codes = ["1309", "600519", "000001", "301392"]
    for code in test_codes:
        is_unlock, info = is_recent_unlock(code, unlock_df)
        status = "是" if is_unlock else "否"
        print(f"股票 {code.zfill(6)} 是否近期解禁: {status}")