Source code for langchain_community.utilities.google_finance

"""调用谷歌财经搜索的工具。"""
from typing import Any, Dict, Optional, cast

from langchain_core.pydantic_v1 import BaseModel, Extra, SecretStr, root_validator
from langchain_core.utils import convert_to_secret_str, get_from_dict_or_env


class GoogleFinanceAPIWrapper(BaseModel):
    """包装器用于SerpApi的Google Finance API

    您可以通过在以下网址注册来创建SerpApi.com密钥：https://serpapi.com/users/sign_up。
    该包装器使用SerpApi.com的python包：
    https://serpapi.com/integrations/python
    要使用，您应该设置环境变量``SERPAPI_API_KEY``
    为您的API密钥，或将`serp_api_key`作为命名参数传递给构造函数。
    示例：
        .. code-block:: python
        from langchain_community.utilities import GoogleFinanceAPIWrapper
        google_Finance = GoogleFinanceAPIWrapper()
        google_Finance.run('langchain')"""

    serp_search_engine: Any
    serp_api_key: Optional[SecretStr] = None

    class Config:
        """此pydantic对象的配置。"""

        extra = Extra.forbid

    @root_validator()
    def validate_environment(cls, values: Dict) -> Dict:
        """验证环境中是否存在API密钥和Python包。"""
        values["serp_api_key"] = convert_to_secret_str(
            get_from_dict_or_env(values, "serp_api_key", "SERPAPI_API_KEY")
        )

        try:
            from serpapi import SerpApiClient

        except ImportError:
            raise ImportError(
                "google-search-results is not installed. "
                "Please install it with `pip install google-search-results"
                ">=2.4.2`"
            )
        serp_search_engine = SerpApiClient
        values["serp_search_engine"] = serp_search_engine

        return values

    def run(self, query: str) -> str:
        """通过Serpapi在Google Finance上运行查询。"""
        serpapi_api_key = cast(SecretStr, self.serp_api_key)
        params = {
            "engine": "google_finance",
            "api_key": serpapi_api_key.get_secret_value(),
            "q": query,
        }

        total_results = {}
        client = self.serp_search_engine(params)
        total_results = client.get_dict()

        if not total_results:
            return "Nothing was found from the query: " + query

        markets = total_results.get("markets", {})
        res = "\nQuery: " + query + "\n"

        if "futures_chain" in total_results:
            futures_chain = total_results.get("futures_chain", [])[0]
            stock = futures_chain["stock"]
            price = futures_chain["price"]
            temp = futures_chain["price_movement"]
            percentage = temp["percentage"]
            movement = temp["movement"]
            res += (
                f"stock: {stock}\n"
                + f"price: {price}\n"
                + f"percentage: {percentage}\n"
                + f"movement: {movement}\n"
            )

        else:
            res += "No summary information\n"

        for key in markets:
            if (key == "us") or (key == "asia") or (key == "europe"):
                res += key
                res += ": price = "
                res += str(markets[key][0]["price"])
                res += ", movement = "
                res += markets[key][0]["price_movement"]["movement"]
                res += "\n"

        return res