Source code for kpler.sdk.resources.supply_demand

from datetime import date
from enum import Enum
from typing import List, Optional

from kpler.sdk import Platform
from kpler.sdk.client import KplerClient
from kpler.sdk.configuration import Configuration
from kpler.sdk.helpers import (
    process_date_parameter,
    process_enum_parameter,
    process_enum_parameters,
    process_list_parameter,
)


[docs]class SupplyDemand(KplerClient): """ The ``SupplyDemand`` endpoint returns supply and demand metrics data values aggregated or split by country for a given product/ zone/ time period/selected snapshot, It also allows listing the available products and snapshots for the main endpoint """ RESOURCE_NAME = "supply-demand" AVAILABLE_PLATFORMS = [Platform.Liquids] def __init__(self, configuration: Configuration, column_ids: bool = True, log_level=None): super().__init__(configuration, self.AVAILABLE_PLATFORMS, column_ids, log_level)
[docs] def get( self, product: Optional[str] = None, metrics: Optional[List[Enum]] = None, split: Optional[Enum] = None, zones: Optional[List[str]] = None, start_date: Optional[date] = None, end_date: Optional[date] = None, snapshot_date: Optional[date] = None, ): """ Args: product: Optional[str] Name of product to retrieve data for metrics: Optional[List[Enum]] ``SupplyDemandMetric`` Returns only provided metrics split: Optional[Enum] ``SupplyDemandSplit`` Choose whether returned data are aggregated or split by country zones: Optional[List[str]] Names of countries/geographical zones start_date: Optional[date] Start of the period (YYYY-MM-DD), must be after 2017-01-01 end_date: Optional[date] End of the period (YYYY-MM-DD), maximum of 18 month from today snapshot_date: Optional[date] Date of the snapshot to retrieve data from (YYYY-MM-DD) Examples: >>> from datetime import date ... from kpler.sdk.resources.supply_demand import SupplyDemand ... from kpler.sdk import SupplyDemandSplit, SupplyDemandMetric ... sd_client = SupplyDemand(config) ... sd_client.get( ... product="Crude/Co", ... metrics=[SupplyDemandMetric.Supply, SupplyDemandMetric.Demand], ... start_date=date(2020,10,1), ... end_date=date(2020,11,1), ... zones=["Japan"], ... split=SupplyDemandSplit.Total ... ) .. csv-table:: :header: "Snapshot Date","Date","Product","Metric","Zones","Value","Unit" "2022-09-15","2020-10-01","Crude/Co",Supply (kbd),"Japan","3","kbd" "2022-09-15","2020-10-01","Crude/Co",Demand (kbd),"Japan","2279","kbd" "2022-09-15","2020-10-01","Crude/Co",Supply (kbd),"Japan","3","kbd" "2022-09-15","2020-10-01","Crude/Co",Demand (kbd),"Japan","2504","kbd" """ query_parameters = { "product": product, "metrics": process_enum_parameters(metrics, to_lower_case=False), "zones": process_list_parameter(zones), "split": process_enum_parameter(split, to_lower_case=False), "startDate": process_date_parameter(start_date), "endDate": process_date_parameter(end_date), "snapshotDate": process_date_parameter(snapshot_date), } return self._get_dataframe(self.RESOURCE_NAME, query_parameters)
[docs] def get_snapshots( self, product: Optional[str] = None, start_date: Optional[date] = None, end_date: Optional[date] = None, ): """ Args: product: Optional[str] Name of product to retrieve snapshots for start_date: Optional[date] Start of the period (YYYY-MM-DD) end_date: Optional[date] End of the period (YYYY-MM-DD) Examples: >>> from datetime import date ... from kpler.sdk.resources.supply_demand import SupplyDemand ... sd_client = SupplyDemand(config) ... sd_client.get_snapshots( ... product="Crude/Co", ... start_date=date(2022,08,1), ... end_date=date(2022,10,1), ... ) .. csv-table:: :header: "Date","Product" "2022-08-14","Crude/Co" "2022-09-01","Crude/Co" """ query_parameters = { "product": product, "startDate": process_date_parameter(start_date), "endDate": process_date_parameter(end_date), } return self._get_dataframe(self.RESOURCE_NAME + "/snapshots", query_parameters)
[docs] def get_products(self): """ Args: None Examples: >>> from datetime import date ... from kpler.sdk.resources.supply_demand import SupplyDemand ... sd_client = SupplyDemand(config) ... sd_client.get_products() .. csv-table:: :header: "Product","metrics","Snapshot Start Date","Snapshot End Date" Crude/Co,"supply,demand,refineryRun,directCrudeUse,balance,netExport,stockChange,balancingFactor",2022-08-15,2022-09-15 """ return self._get_dataframe(self.RESOURCE_NAME + "/products", params={})