用数据说话 - 港股打新有多赚？

def get_ipo_price_from_ai(stock_code: str, stock_name: str, max_retries: int = 3) -> float:
    """通过 qwen3-max 模型获取股票的 IPO 发行价"""
    client = OpenAI(
        api_key=os.getenv("QWEN_API_KEY"),
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    )
    
    for attempt in range(1, max_retries + 1):
        try:
            completion = client.chat.completions.create(
                model="qwen3-max",
                messages=[
                    {"role": "user", "content": f"{stock_code} {stock_name} 港股 IPO 发行价格，直接返回数字，无需单位"},
                ],
                extra_body={"enable_search": True}
            )
            
            content = completion.choices[0].message.content.strip()
            
            # 检查返回是否为纯数字
            if re.match(r'^\d+\.?\d*$', content):
                price = float(content)
                return price
            
            # 尝试提取数字
            match = re.search(r'(\d+\.?\d*)', content)
            if match:
                price = float(match.group(1))
                if attempt < max_retries:
                    continue  # 重试以获取更好的结果
                return price
                
        except Exception as e:
            if attempt < max_retries:
                continue
    
    return 0.0  # 失败返回0

def fetch_ipo_prices_concurrent(stock_list: pd.DataFrame, max_workers: int = 10) -> pd.DataFrame:
    """并发获取股票的 IPO 发行价"""
    result = stock_list.copy()
    result['ipo_price'] = 0.0
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_idx = {
            executor.submit(get_ipo_price_from_ai, row['code'], row['name']): idx
            for idx, row in result.iterrows()
        }
        
        for future in as_completed(future_to_idx):
            idx = future_to_idx[future]
            price = future.result()
            result.at[idx, 'ipo_price'] = price
    
    # 过滤掉价格为0的股票
    result = result[result['ipo_price'] > 0].copy()
    return result