Added price_performance.py

Author: shugaoye

openbb_akshare/__init__.py

@@ -14,6 +14,7 @@
 from openbb_akshare.models.equity_profile import AKShareEquityProfileFetcher
 from openbb_akshare.models.historical_dividends import AKShareHistoricalDividendsFetcher
 from openbb_akshare.models.income_statement import AKShareIncomeStatementFetcher
+from openbb_akshare.models.price_performance import AKSharePricePerformanceFetcher
 
 # mypy: disable-error-code="list-item"
 
@@ -38,5 +39,6 @@
         "EquityInfo": AKShareEquityProfileFetcher,
         "HistoricalDividends": AKShareHistoricalDividendsFetcher,
         "IncomeStatement": AKShareIncomeStatementFetcher,
+        "PricePerformance": AKSharePricePerformanceFetcher,
     }
 )

openbb_akshare/models/price_performance.py

@@ -0,0 +1,95 @@
+"""AKShare Price Performance Model."""
+
+# pylint: disable=unused-argument
+from typing import Any, Dict, List, Optional
+from warnings import warn
+
+from openbb_core.provider.abstract.fetcher import Fetcher
+from openbb_core.provider.standard_models.recent_performance import (
+    RecentPerformanceData,
+    RecentPerformanceQueryParams,
+)
+from openbb_fmp.utils.helpers import create_url, get_data_urls
+from pydantic import Field, model_validator
+
+
+class AKSharePricePerformanceQueryParams(RecentPerformanceQueryParams):
+    """AKShare Price Performance Query.
+
+    Source: https://akshare.akfamily.xyz/data/index/index.html#id3
+    """
+
+    __json_schema_extra__ = {"symbol": {"multiple_items_allowed": True}}
+
+
+class AKSharePricePerformanceData(RecentPerformanceData):
+    """AKShare Price Performance Data."""
+
+    #symbol: str = Field(description="The ticker symbol.")
+
+    __alias_dict__ = {
+        "one_day": "1D",
+        "one_week": "5D",
+        "one_month": "1M",
+        "three_month": "3M",
+        "six_month": "6M",
+        "one_year": "1Y",
+        "three_year": "3Y",
+        "five_year": "5Y",
+        "ten_year": "10Y",
+    }
+
+    @model_validator(mode="before")
+    @classmethod
+    def replace_zero(cls, values):  # pylint: disable=no-self-argument
+        """Replace zero with None and convert percents to normalized values."""
+        if isinstance(values, dict):
+            for k, v in values.items():
+                if k != "symbol":
+                    values[k] = None if v == 0 else float(v) / 100
+        return values
+
+
+class AKSharePricePerformanceFetcher(
+    Fetcher[
+        AKSharePricePerformanceQueryParams,
+        List[AKSharePricePerformanceData],
+    ]
+):
+    """Transform the query, extract and transform the data from the AKShare endpoints."""
+
+    @staticmethod
+    def transform_query(params: Dict[str, Any]) -> AKSharePricePerformanceQueryParams:
+        """Transform the query params."""
+        return AKSharePricePerformanceQueryParams(**params)
+
+    @staticmethod
+    async def extract_data(
+        query: AKSharePricePerformanceQueryParams,
+        credentials: Optional[Dict[str, str]],
+        **kwargs: Any,
+    ) -> List[Dict]:
+        """Return the raw data from the AKShare endpoint."""
+        import pandas as pd
+        symbols = query.symbol.split(",")
+
+        def get_data(code: str) -> dict:
+            from mysharelib.tools import calculate_price_performance, last_closing_day
+            from openbb_akshare.utils.helpers import get_list_date, ak_download
+
+            df = ak_download(symbol=code, start_date=get_list_date(code), end_date=last_closing_day())
+            df['date'] = pd.to_datetime(df['date'])
+            df.set_index('date', inplace=True)
+            return calculate_price_performance(code, df)
+
+        return [get_data(symbol) for symbol in symbols]
+
+    @staticmethod
+    def transform_data(
+        query: AKSharePricePerformanceQueryParams,
+        data: List[Dict],
+        **kwargs: Any,
+    ) -> List[AKSharePricePerformanceData]:
+        """Return the transformed data."""
+
+        return [AKSharePricePerformanceData.model_validate(i) for i in data]

openbb_akshare/utils/fetch_equity_info.py

@@ -60,7 +60,7 @@ def fetch_equity_info(symbol: str, use_cache: bool = True) -> pd.DataFrame:
         data = cache.read_dataframe()
         result = data[data["symbol"] == symbol_f]
         if not result.empty:
-            logger.info(f"Using cached equity info for symbol: {symbol_f}")
+            #logger.info(f"Using cached equity info for symbol: {symbol_f}")
             return result
 
     columns = list(EQUITY_INFO_SCHEMA.keys())
@@ -112,6 +112,6 @@ def fetch_equity_info(symbol: str, use_cache: bool = True) -> pd.DataFrame:
     result = data[data["symbol"] == symbol_f]
     result["listed_date"] = pd.to_datetime(result["listed_date"], unit='ms')
     result["established_date"] = pd.to_datetime(result["established_date"], unit='ms')
-    logger.info(f"Fetched equity info for symbol: {symbol_f}")
+    # logger.info(f"Fetched equity info for symbol: {symbol_f}")
 
     return result

openbb_akshare/utils/helpers.py

@@ -54,8 +54,8 @@ def get_list_date(symbol: str) -> dateType:
 
 def check_cache(symbol: str, 
         cache: TableCache,
-        api_key : str = "",
-        period: str = "daily"
+        api_key : Optional[str] = "",
+        period: Optional[str] = "daily"
         ) -> bool:
     """
     Check if the cache contains the latest data for the given symbol.
@@ -79,10 +79,10 @@ def ak_download_without_cache(
         symbol: str,
         start_date: str,
         end_date: str,
-        period: str = "daily",
-        use_cache: bool = True, 
-        api_key : str = "",
-        adjust: str = "",
+        period: Optional[str] = "daily",
+        use_cache: Optional[bool] = True, 
+        api_key : Optional[str] = "",
+        adjust: Optional[str] = "",
     ) -> DataFrame:
     """
     Downloads historical equity data without using cache.
@@ -119,22 +119,25 @@ def ak_download_without_cache(
 
 def ak_download(
         symbol: str,
-        start_date: dateType,
-        end_date: dateType,
-        period: str = "daily",
+        start_date: Optional[dateType] = None,
+        end_date: Optional[dateType] = None,
+        period: Optional[str] = "daily",
         use_cache: Optional[bool] = True,
-        adjust: str = "",
+        adjust: Optional[str] = "",
     ) -> DataFrame:
 
     from mysharelib.tools import get_valid_date
 
+    if start_date is None:
+        start_date = (datetime.now() - timedelta(days=365)).date()
+    if end_date is None: end_date = datetime.now().date()
+    start_dt = get_valid_date(start_date)
+    end_dt = get_valid_date(end_date)
+
     # Retrieve data from cache first
     symbol_b, symbol_f, market = normalize_symbol(symbol)
     cache = TableCache(EQUITY_HISTORY_SCHEMA, project=project_name, table_name=f"{market}{symbol_b}", primary_key="date")
 
-    start_dt = get_valid_date(start_date)
-    end_dt = get_valid_date(end_date)
-
     if use_cache:
         check_cache(symbol=symbol_b, cache=cache, api_key="", period=period)
         data_from_cache = cache.fetch_date_range(start_dt.strftime("%Y-%m-%d"), end_dt.strftime("%Y-%m-%d"))

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "openbb-akshare"
-version = "0.5.7"
+version = "0.5.9"
 description = "AKShare extension for OpenBB"
 authors = ["Roger Ye <shugaoye@yahoo.com>"]
 license = "AGPL-3.0-only"

tests/test_helpers.py

@@ -50,3 +50,18 @@ def test_ak_download(logger):
     logger.info(f"Downloading data for {symbol} from {start_date} to {end_date}")
     df = ak_download(symbol, start_date, end_date)
     assert not df.empty
+
+def test_price_performance(logger):
+    def get_data(code: str) -> dict:
+        import pandas as pd
+        from mysharelib.tools import calculate_price_performance, last_closing_day
+        from openbb_akshare.utils.helpers import get_list_date, ak_download
+
+        df = ak_download(symbol=code, start_date=get_list_date(code), end_date=last_closing_day())
+        df['date'] = pd.to_datetime(df['date'])
+        df.set_index('date', inplace=True)
+        return calculate_price_performance(code, df)
+
+    symbols = '601919,600028'
+    symbols = symbols.split(",")
+    logger.info([get_data(symbol) for symbol in symbols])