mindspore/utils.py

from dataclasses import dataclass, field
from typing import List, Dict, Optional
import pandas as pd
import asyncio
import requests
from bs4 import BeautifulSoup
from datetime import datetime, timedelta
import tushare as ts
import json

# 使用你的Tushare API Token
ts.set_token('')
pro = ts.pro_api()

# 加载配置文件
class Config:
    with open('config.json', 'r') as f:
        config_data = json.load(f)
    LLM_API_KEY = ''  # 替换为你的API Key
    LLM_URL = 'https://api.deepseek.com/chat/completions'
    LLM_MODEL = 'deepseek-chat'

@dataclass
class TickerClass:
    name: str  # 股票代码
    display_name: str = field(default="")  # 股票名称
    hist_data: Optional[pd.DataFrame] = field(default=None)  # 历史数据
    balance_sheet: Optional[pd.DataFrame] = field(default=None)  # 资产负债表
    financials: Optional[pd.DataFrame] = field(default=None)  # 财务数据
    news: Optional[Dict] = field(default=None)  # 新闻数据
    analyst_ratings: str = field(default=None)  # 分析师评级
    price: float = field(default=None)  # 股票当前价格
    sentiment_analysis: str = field(default=None)  # 情感分析
    industry_analysis: str = field(default=None)  # 行业分析
    final_analysis: str = field(default=None)  # 最终的分析


    
# 获取文章的内容，输入是文章的URL，输出是文章的文本
def get_article_text(url: str) -> str:
    try:
        response = requests.get(url)  
        soup = BeautifulSoup(response.content, 'html.parser')  
        article_text = ' '.join([p.get_text() for p in soup.find_all('p')]) 
        return article_text
    except:
        return "Error retrieving article text."

# 获取股票数据
def get_stock_data(ticker: str, days: int = 30) -> tuple:
    end_date = datetime.now().strftime('%Y%m%d')
    start_date = (datetime.now() - timedelta(days=days)).strftime('%Y%m%d')

    # 获取日线行情数据
    hist_data = pro.daily(ts_code=ticker, start_date=start_date, end_date=end_date)
    
    # 获取资产负债表
    balance_sheet = pro.balancesheet(ts_code=ticker)
    
    # 获取利润表
    financials = pro.income(ts_code=ticker)
    
    # Tushare 不提供类似 yfinance 的新闻接口，这里返回空字典
    news = {}

    return hist_data, balance_sheet, financials, news

# 获取股票名称
def get_stock_name(ticker: str) -> str:
    stock_info = pro.stock_basic(ts_code=ticker, fields='name')
    if not stock_info.empty:
        return stock_info.iloc[0]['name']
    return "Unknown"


# 获取分析师预测数据
def get_analyst_ratings(ticker: str) -> str:
    forecast = pro.forecast(ts_code=ticker)
    if forecast.empty:
        return "No earnings forecast available."
    
    latest_forecast = forecast.iloc[0]
    forecast_summary = f"Latest earnings forecast for {ticker}:\n {str(latest_forecast.to_dict())}"
    
    return forecast_summary

# 获取当前股票价格
def get_current_price(ticker: str) -> float:
    df = pro.daily(ts_code=ticker)
    if df.empty:
        return None
    return df.iloc[0]['close']

import requests

# 分析情感数据，结合新闻、历史数据、财务数据等，并让大模型给出情绪分数
async def get_sentiment_analysis(news_title: str, news_content: str, related_stocks: List[str]) -> Dict[str, Dict]:
    print(f"Analyzing sentiment for the news: {news_title}")

    news_analysis_results = {}

    for stock in related_stocks:
   
        hist_data, balance_sheet, financials, _ = get_stock_data(stock)

        # 构建要发送给大模型的 prompt
        messages = [
            {
                "role": "system", 
                "content": (
                    "你是一名专业的金融分析师，专注于金融新闻和财务数据的情感分析。"
                    "你将根据新闻内容、股票历史数据、财务数据和资产负债表等信息，分析该新闻对相关股票的情绪影响，"
                    "并给出情绪分数（0到1之间，1表示非常积极，0表示非常消极），以及投资建议（例如：买入、持有、卖出）。"
                    "请详细考虑市场反应、公司财务健康状况和未来前景。"
                )
            },
            {
                "role": "user", 
                "content": (
                    f"以下是关于股票 {stock} 的分析请求。\n\n"
                    f"新闻标题: {news_title}\n"
                    f"新闻内容: {news_content}\n\n"
                    f"股票历史数据 (过去一年): {hist_data.to_dict()}\n\n"
                    f"财务数据: {financials.to_dict()}\n\n"
                    f"资产负债表: {balance_sheet.to_dict()}\n\n"
                    f"请基于以上信息给出以下分析：\n"
                    f"1. 该新闻对股票 {stock} 的情绪分数（0-1，1 表示非常积极）\n"
                    f"2. 对股票的投资建议（买入、持有或卖出），并详细说明原因。"
                )
            }
        ]

        headers = {
            "Authorization": f"Bearer {Config.LLM_API_KEY}",
            "Content-Type": "application/json"
        }

        # 请求大模型 API 进行分析
        response = requests.post(Config.LLM_URL, json={"model": Config.LLM_MODEL, "messages": messages}, headers=headers)

        if response.status_code == 200:
            
            sentiment_result = response.json()['choices'][0]['message']['content']

            
            sentiment_score = extract_sentiment_score(sentiment_result)
            investment_suggestion = extract_investment_suggestion(sentiment_result)

            
            news_analysis_results[stock] = {
                "情绪分数": sentiment_score,
                "投资建议": investment_suggestion,
                "依据": sentiment_result
            }
        else:
            news_analysis_results[stock] = {
                "情绪分数": "无法获取",
                "投资建议": "错误",
                "依据": "分析过程中出错"
            }

    return news_analysis_results

# 提取情感分数的辅助函数
def extract_sentiment_score(sentiment_text: str) -> float:
    import re
    match = re.search(r"情绪分数[:：]\s*(\d*\.?\d+)", sentiment_text)
    if match:
        return float(match.group(1))
    return 0.5  

# 提取投资建议的辅助函数
def extract_investment_suggestion(sentiment_text: str) -> str:
    import re
    match = re.search(r"投资建议[:：]\s*(买入|持有|卖出)", sentiment_text)
    if match:
        return match.group(1)
    return "持有"  


# 调用大模型进行行业分析
async def get_industry_analysis(ticker: TickerClass) -> None:
    print(f"Industry analysis for {ticker.name}")
    stock_info = pro.stock_basic(ts_code=ticker.name, fields='industry')

    if stock_info.empty:
        industry = "未知"
    else:
        industry = stock_info.iloc[0]['industry']

    payload = {
        "model": Config.LLM_MODEL,
        "messages": [
            {"role": "system", "content": f"你是一个行业分析助手。请为 {ticker.name} 提供行业分析。"},
            {"role": "user", "content": f"行业: {industry}\n请分析该行业的趋势、增长前景、监管变化和竞争格局。"}
        ]
    }

    headers = {
        "Authorization": f"Bearer {Config.LLM_API_KEY}",
        "Content-Type": "application/json"
    }

    response = requests.post(Config.LLM_URL, json=payload, headers=headers)

    if response.status_code == 200:
        ticker.industry_analysis = response.json()['choices'][0]['message']['content']
    else:
        ticker.industry_analysis = "行业分析出错"



# 调用大模型进行最终分析
async def get_final_analysis(ticker: TickerClass) -> None:
    print(f"Final analysis for {ticker.name}")
    
    # 在构建的prompt中，明确要求模型引用情绪分析、财务数据和历史数据，并提供详细的依据
    payload = {
        "model": Config.LLM_MODEL,
        "messages": [
            {
                "role": "system", 
                "content": (
                    f"你是一位专业的金融分析师，负责提供股票 {ticker.name} 的最终投资建议。"
                    "请基于以下内容，给出详细的建议并明确引用数据：\n"
                    f"1. 情绪分析: {ticker.sentiment_analysis}\n"
                    f"2. 最新分析师评级: {ticker.analyst_ratings}\n"
                    f"3. 行业分析: {ticker.industry_analysis}\n"
                    "你需要综合公司财务健康状况、市场表现、行业前景和情绪分析，"
                    "并明确说明你的建议是基于哪些具体数据、分析或情绪影响，"
                    "最终给出是否建议买入、持有或卖出股票。"
                )
            },
            {
                "role": "user", 
                "content": (
                    f"股票: {ticker.name}\n\n"
                    f"请基于以上内容提供详细的投资建议，并引用财务数据、市场趋势、竞争地位和潜在风险等作为依据。"
                )
            }
        ]
    }

    headers = {
        "Authorization": f"Bearer {Config.LLM_API_KEY}",
        "Content-Type": "application/json"
    }

    # 发送请求，获取大模型生成的结果
    response = requests.post(Config.LLM_URL, json=payload, headers=headers)

    if response.status_code == 200:
        ticker.final_analysis = response.json()['choices'][0]['message']['content']
    else:
        ticker.final_analysis = "生成最终分析出错"


# 根据所有股票的分析，进行股票排名
def rank_companies(ticker_info_list: List[TickerClass], industry: str) -> str:
    print(f"Ranking ...")

    # 汇总所有股票的详细分析，并将其作为投资建议的依据，特别强调情绪分析
    analysis_text = "\n\n".join(
        f"股票: {ticker.name} - {ticker.display_name}\n"  # 输出股票代码和名称
        f"当前价格: {ticker.price}\n"
        f"新闻情绪分析: {ticker.sentiment_analysis}\n"
        f"最新分析师评级: {ticker.analyst_ratings}\n"
        f"行业分析: {ticker.industry_analysis}\n"
        f"最终投资建议: {ticker.final_analysis}" 
        for ticker in ticker_info_list
    )

    # 构建详细的提示，强调新闻情绪分析的影响，并要求模型按照指定模板输出结果
    payload = {
        "model": Config.LLM_MODEL,
        "messages": [
            {
                "role": "system", 
                "content": (
                    "你是一位资深的金融分析师，负责为房地产行业中的公司按投资潜力进行排名。"
                    "在这次分析中，市场新闻的情绪分析是至关重要的判断依据，"
                    "请务必将新闻情绪分析的结果作为每个股票排名和投资建议的核心因素之一，"
                    "并结合公司的财务状况、市场趋势和竞争地位进行综合分析。"
                    "请确保分析结果按照以下格式输出，并且每次都严格遵循以下示例格式：\n"
                    "### 投资吸引力排名\n"
                    "1. **000560.SZ** - **股票名称**\n"
                    "   **理由**:\n"
                    "   **新闻情绪分析**: 新闻情绪的详细分析结果及对市场反应的预期。\n"
                    "   **财务健康状况**: 描述财务状况。\n"
                    "   **市场趋势**: 描述市场趋势。\n"
                    "   **竞争地位**: 描述竞争地位。\n"
                    "   **潜在风险**: 描述潜在风险。\n"
                    "   **建议**: 买入或卖出。\n"
                    "2. **002016.SZ** - **股票名称**\n"
                    "   **理由**:\n"
                    "   **新闻情绪分析**: 新闻情绪的详细分析结果及对市场反应的预期。\n"
                    "   **财务健康状况**: 描述财务状况。\n"
                    "   **市场趋势**: 描述市场趋势。\n"
                    "   **竞争地位**: 描述竞争地位。\n"
                    "   **潜在风险**: 描述潜在风险。\n"
                    "   **建议**: 买入或卖出。\n"
                    "3. **000736.SZ** - **股票名称**\n"
                    "   **理由**:\n"
                    "   **新闻情绪分析**: 新闻情绪的详细分析结果及对市场反应的预期。\n"
                    "   **财务健康状况**: 描述财务状况。\n"
                    "   **市场趋势**: 描述市场趋势。\n"
                    "   **竞争地位**: 描述竞争地位。\n"
                    "   **潜在风险**: 描述潜在风险。\n"
                    "   **建议**: 买入或卖出。\n"
                    "# 总结\n"
                    "最后进行总结，说明哪个股票最具投资吸引力，并解释新闻情绪分析对其的影响。"
                )
            },
            {
                "role": "user", 
                "content": (
                    f"行业: {industry}\n\n"
                    f"公司分析:\n{analysis_text}\n\n"
                    f"请根据提供的分析，将这些公司按投资吸引力进行排名，并提供详细的理由。"
                    "请务必明确引用新闻情绪分析结果，结合财务健康状况和市场趋势等作为排名依据。"
                )
            }
        ]
    }

    headers = {
        "Authorization": f"Bearer {Config.LLM_API_KEY}",
        "Content-Type": "application/json"
    }

    # 调用模型API进行排名分析
    response = requests.post(Config.LLM_URL, json=payload, headers=headers)

    if response.status_code == 200:
        return response.json()['choices'][0]['message']['content']
    else:
        return "公司排名出错"





async def get_openai_verdict_2(news_title: str, news_content: str, tickers: List[str], industry: str):
    ticker_info_list = []

    # 遍历股票代码，获取股票信息
    for ticker in tickers:
        stock_name = get_stock_name(ticker)  # 获取股票名称
        temp_ticker_info = TickerClass(name=ticker, display_name=stock_name)
        temp_ticker_info.hist_data, temp_ticker_info.balance_sheet, temp_ticker_info.financials, temp_ticker_info.news = get_stock_data(ticker, days=30)
        temp_ticker_info.analyst_ratings = get_analyst_ratings(ticker)
        temp_ticker_info.price = get_current_price(ticker)
        ticker_info_list.append(temp_ticker_info)

    # 执行情感分析，结合手动输入的新闻标题和内容
    sentiment_results = await get_sentiment_analysis(news_title, news_content, tickers)

    # 将情感分析结果赋值给每个股票对象
    for ticker_object in ticker_info_list:
        if ticker_object.name in sentiment_results:
            ticker_object.sentiment_analysis = sentiment_results[ticker_object.name]["依据"]

    # 异步调用行业分析
    tasks2 = [get_industry_analysis(ticker_object) for ticker_object in ticker_info_list]
    await asyncio.gather(*tasks2)

    # 异步调用最终分析
    tasks3 = [get_final_analysis(ticker_object) for ticker_object in ticker_info_list]
    await asyncio.gather(*tasks3)

    # 最后对所有股票进行排名并返回
    final_ranking = rank_companies(ticker_info_list, industry)
    return final_ranking

'''
async def main():
    # 手动输入的新闻标题和新闻内容
    news_title = "央行降低存量房贷利率，多重政策推动房地产市场回暖"
    news_content = (
        "2024年9月24日，央行宣布多项措施以提振房地产市场，包括降低存量房贷利率、统一房贷最低首付比例、"
        "引导LPR（贷款市场报价利率）下行等。这些政策旨在减轻购房者负担，增加市场需求，并提振整体经济。"
        "消息公布后，房地产股大幅上涨，多只港股内房股涨幅超过10%。"
    )
    
    # 相关股票代码
    tickers = ['000560.SZ', '000736.SZ', '002016.SZ']  
    industry = "房地产"

    # 执行情绪分析和投资建议生成
    final_rankings = await get_openai_verdict_2(news_title, news_content, tickers=tickers, industry=industry)

    # 打印最终排名结果
    print(final_rankings)

# 运行主函数
if __name__ == "__main__":
    asyncio.run(main())

'''