Source code for kpler.sdk.resources.trades_snapshot

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

from pandas import DataFrame

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_parameters,
    process_list_parameter,
    process_snapshot_date_parameter,
)


[docs]class TradesSnapshot(KplerClient): """ The ``TradesSnapshot`` query returns a snapshot of our trades data at a specified time. """ RESOURCE_NAME = "trades/snapshot" COLUMNS_RESOURCE_NAME = "trades" AVAILABLE_PLATFORMS = [Platform.Dry, Platform.Liquids, Platform.LNG, Platform.LPG] 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_columns(self) -> DataFrame: """ This endpoint Provides the last snapshot of all trades. The snapshot is a result of getTrades for all zones, vessels, and products, overall the history.. Examples: >>> from kpler.sdk.resources.trades_snapshot import TradesSnapshot ... trades_snapshot_client = TradesSnapshot(config) ... trades_snapshot_client.get_columns() .. csv-table:: :header: "id","name","description","deprecated","type" "vessel_name","Vessel","Name of the vessel","False","string" "start","Date (origin)","Departure date of the vessel","False","datetime yyyy-MM-dd HH:mm" "origin_location_name","Origin","Origin location of the cargo","False","string" "origin_eta_source","Eta source (origin)","Source of the Estimated Time of Arrival to the Installation of Origin information (Port, Analyst, etc.)","False","string" "cargo_origin_cubic_meters","Volume (origin m3)","None","False","long" "...","...","...","...","..." """ return self._get_columns_for_resource(self.COLUMNS_RESOURCE_NAME)
[docs] def get( self, snapshot_date: Optional[date] = None, vessels: Optional[List[str]] = None, from_installations: Optional[List[str]] = None, to_installations: Optional[List[str]] = None, from_zones: Optional[List[str]] = None, to_zones: Optional[List[str]] = None, buyers: Optional[List[str]] = None, sellers: Optional[List[str]] = None, products: Optional[List[str]] = None, start_date: Optional[date] = None, end_date: Optional[date] = None, trade_status: Optional[List[Enum]] = None, columns: Optional[List[str]] = None, ) -> DataFrame: """ Args: snapshot_date: Optional[date] Date of the snapshot, if empty it returns the most recent snapshot vessels: Optional[List[str]] Names/IMO's of vessels from_installations: Optional[List[str]] Names of the origin installations to_installations: Optional[List[str]] Names of the destination installations (terminal/refinery) from_zones: Optional[List[str]] Names of the origin zones (port/region/country/continent) to_zones: Optional[List[str]] Names of the destination zones (port/region/country/continent) buyers: Optional[List[str]] Buyers of the cargo sellers: Optional[List[str]] Sellers of the cargo products: Optional[List[str]] Names of products start_date: Optional[date] Start of the period end_date: Optional[date] End of the period trade_status: Optional[List[Enum]] ``TradesStatus`` Return only trades of a particular status. By default value is scheduled. columns: Optional[List[str]] Retrieve all available columns when set to "all" Examples: >>> from datetime import date, timedelta, datetime ... from kpler.sdk.resources.trades_snapshot import TradesSnapshot ... trades_snapshot_client = TradesSnapshot(config) ... snapshots = trades_snapshot_client.get( ... snapshot_date = datetime.strptime('2022/01/01', '%Y/%m/%d'), ... to_zones = ["France"], ... products = ["crude"], ... start_date = datetime.strptime('2021/01/01', '%Y/%m/%d'), ... columns=[ ... "vessel_name", ... "origin_country_name", ... "origin_location_name", ... "destination_location_name", ... "start", ... "end" ... ] ... ) >>> print(snapshots.headers['snapshotName']) "2022-01-01T000100" >>> print(snapshots.headers['lastAvailableDate']) "2022-10-19T034600" .. csv-table:: :header: "vessel_name", "origin_country_name", "origin_location_name","destination_location_name","start","end","trade_id","product_id" "Sakura Princess", "Russian Federation", "Novorossiysk", "Fos", "2022-02-05 18:48:00", "2022-02-18 23:36:00", "13163188", "1360" "Alfa Baltica", "United Kingdom", "Hound Point", "Le Havre", "2022-01-29 08:31:00", "2022-02-09 22:25:00", "13214389", "1502" "Scf Baikal", "Equatorial Guinea", "Zafiro FPSO", "Fos", "2022-01-27 14:18:00", "2022-02-20 02:17:00", "13263935", "2430" "Calida", "Russian Federation", "Umba FSO", "Le Havre", "2022-01-25 21:59:00", "2022-02-02 11:00:00", "13107517", "1940" "Siri Knutsen", "United Kingdom", "Culzean", "Le Havre", "2022-01-25 18:50:00", "2022-01-27 17:01:00", "13208418", "2488" "...", "...", "...", "...", "...", "...", "...", "..." """ query_parameters: Dict[str, Optional[Any]] = { "snapshotDate": process_date_parameter(snapshot_date), "vessels": process_list_parameter(vessels), "fromInstallations": process_list_parameter(from_installations), "toInstallations": process_list_parameter(to_installations), "fromZones": process_list_parameter(from_zones), "toZones": process_list_parameter(to_zones), "buyers": process_list_parameter(buyers), "sellers": process_list_parameter(sellers), "products": process_list_parameter(products), "startDate": process_snapshot_date_parameter(start_date), "endDate": process_snapshot_date_parameter(end_date), "tradeStatus": process_enum_parameters(trade_status), "columns": process_list_parameter(columns), } header_parameters: List[str] = ["snapshotName", "lastAvailableDate"] return self._get_dataframe(self.RESOURCE_NAME, query_parameters, header_parameters)