{ "cells": [ { "cell_type": "markdown", "id": "92475c69", "metadata": {}, "source": [ "# Spark Analytics Gallery - Hubs Cargo USGC Swaps vs Spark30S\n", "\n", "Here we plot USGC Swaps data (Hubs Cargo API) vs Spark30S Atlantic freight rates.\n", "\n", "For a full explanation of how to import our Hubs Cargo or Spark30S data, please refer to our Python Jupyter Notebook Code Samples:\n", "\n", "https://www.sparkcommodities.com/api/code-examples/jupyter.html\n", "\n", "### Have any questions?\n", "\n", "If you have any questions regarding our API, or need help accessing specific datasets, please contact us at:\n", "\n", "__data@sparkcommodities.com__\n", "\n", "or refer to our API website for more information about this endpoint:\n", "https://www.sparkcommodities.com/api/" ] }, { "cell_type": "markdown", "id": "c5716130", "metadata": {}, "source": [ "## 1. Importing Data\n", "\n", "Here we define the functions that allow us to retrieve the valid credentials to access the Spark API.\n", "\n", "__This section can remain unchanged for most Spark API users.__" ] }, { "cell_type": "code", "execution_count": null, "id": "38f5c192", "metadata": {}, "outputs": [], "source": [ "import json\n", "import os\n", "import sys\n", "import pandas as pd\n", "import numpy as np\n", "from base64 import b64encode\n", "from urllib.parse import urljoin\n", "from io import StringIO\n", "import datetime\n", "\n", "try:\n", " from urllib import request, parse\n", " from urllib.error import HTTPError\n", "except ImportError:\n", " raise RuntimeError(\"Python 3 required\")" ] }, { "cell_type": "code", "execution_count": null, "id": "33fb0640", "metadata": {}, "outputs": [], "source": [ "API_BASE_URL = \"https://api.sparkcommodities.com\"\n", "\n", "\n", "def retrieve_credentials(file_path=None):\n", " \"\"\"\n", " Find credentials either by reading the client_credentials file or reading\n", " environment variables\n", " \"\"\"\n", " if file_path is None:\n", " client_id = os.getenv(\"SPARK_CLIENT_ID\")\n", " client_secret = os.getenv(\"SPARK_CLIENT_SECRET\")\n", " if not client_id or not client_secret:\n", " raise RuntimeError(\n", " \"SPARK_CLIENT_ID and SPARK_CLIENT_SECRET environment vars required\"\n", " )\n", " else:\n", " # Parse the file\n", " if not os.path.isfile(file_path):\n", " raise RuntimeError(\"The file {} doesn't exist\".format(file_path))\n", "\n", " with open(file_path) as fp:\n", " lines = [l.replace(\"\\n\", \"\") for l in fp.readlines()]\n", "\n", " if lines[0] in (\"clientId,clientSecret\", \"client_id,client_secret\"):\n", " client_id, client_secret = lines[1].split(\",\")\n", " else:\n", " print(\"First line read: '{}'\".format(lines[0]))\n", " raise RuntimeError(\n", " \"The specified file {} doesn't look like to be a Spark API client \"\n", " \"credentials file\".format(file_path)\n", " )\n", "\n", " print(\">>>> Found credentials!\")\n", " \n", " return client_id, client_secret\n", "\n", "\n", "def do_api_post_query(uri, body, headers):\n", " \"\"\"\n", " OAuth2 authentication requires a POST request with client credentials before accessing the API.\n", " This POST request will return an Access Token which will be used for the API GET request.\n", " \"\"\"\n", " url = urljoin(API_BASE_URL, uri)\n", "\n", " data = json.dumps(body).encode(\"utf-8\")\n", "\n", " # HTTP POST request\n", " req = request.Request(url, data=data, headers=headers)\n", " try:\n", " response = request.urlopen(req)\n", " except HTTPError as e:\n", " print(\"HTTP Error: \", e.code)\n", " print(e.read())\n", " sys.exit(1)\n", "\n", " resp_content = response.read()\n", "\n", " # The server must return HTTP 201. Raise an error if this is not the case\n", " assert response.status == 201, resp_content\n", "\n", " # The server returned a JSON response\n", " content = json.loads(resp_content)\n", "\n", " return content\n", "\n", "\n", "def do_api_get_query(uri, access_token, format='json'):\n", " \"\"\"\n", " After receiving an Access Token, we can request information from the API.\n", " Supports both JSON (default) and CSV responses.\n", " \"\"\"\n", " url = urljoin(API_BASE_URL, uri)\n", "\n", " if format == 'json':\n", " headers = {\n", " \"Authorization\": \"Bearer {}\".format(access_token),\n", " \"Accept\": \"application/json\",\n", " }\n", " elif format == 'csv':\n", " headers = {\n", " \"Authorization\": \"Bearer {}\".format(access_token),\n", " \"Accept\": \"text/csv\",\n", " }\n", " else:\n", " raise ValueError(\"The format parameter only takes 'csv' or 'json' as inputs\")\n", "\n", " print(\"Fetching {}\".format(url))\n", "\n", " # HTTP GET request\n", " req = request.Request(url, headers=headers)\n", " try:\n", " response = request.urlopen(req)\n", " except HTTPError as e:\n", " print(\"HTTP Error: \", e.code)\n", " print(e.read())\n", " sys.exit(1)\n", "\n", " resp_content = response.read()\n", "\n", " # The server must return HTTP 200. Raise an error if this is not the case\n", " assert response.status == 200, resp_content\n", "\n", " # Storing response based on requested format\n", " if format == 'json':\n", " content = json.loads(resp_content)\n", " elif format == 'csv':\n", " content = resp_content\n", "\n", " return content\n", "\n", "\n", "def get_access_token(client_id, client_secret):\n", "\n", " payload = \"{}:{}\".format(client_id, client_secret).encode()\n", " headers = {\n", " \"Authorization\": b64encode(payload).decode(),\n", " \"Accept\": \"application/json\",\n", " \"Content-Type\": \"application/json\",\n", " }\n", " body = {\n", " \"grantType\": \"clientCredentials\",\n", " }\n", "\n", " content = do_api_post_query(uri=\"/oauth/token/\", body=body, headers=headers)\n", "\n", " print(\n", " \">>>> Successfully fetched an access token {}****, valid {} seconds.\".format(\n", " content[\"accessToken\"][:5], content[\"expiresIn\"]\n", " )\n", " )\n", "\n", " return content[\"accessToken\"]" ] }, { "cell_type": "markdown", "id": "fd3171a8", "metadata": {}, "source": [ "## N.B. Credentials\n", "\n", "Here we call the above functions, and input the file path to our credentials.\n", "\n", "N.B. You must have downloaded your client credentials CSV file before proceeding. Please refer to the API documentation if you have not dowloaded them already.\n", "\n", "The code then prints the available prices that are callable from the API, and their corresponding Python ticker names are displayed as a list at the bottom of the Output." ] }, { "cell_type": "code", "execution_count": null, "id": "fd7e89bf", "metadata": {}, "outputs": [], "source": [ "# Insert file path to your client credentials here\n", "client_id, client_secret = retrieve_credentials(file_path=\"/tmp/client_credentials.csv\")\n", "\n", "# Authenticate:\n", "access_token = get_access_token(client_id, client_secret)" ] }, { "cell_type": "markdown", "id": "0994ce16", "metadata": {}, "source": [ "# 2. USGC Swaps Data\n", "\n", "Here we define the functions used to call the USGC Swaps data from the Hubs Cargo API. For more information on the Hubs Cargo API, please visit the API website or download our Hubs Cargo code sample:\n", "\n", "https://www.sparkcommodities.com/api/hubs/lng-cargo.html" ] }, { "cell_type": "code", "execution_count": null, "id": "dff2524b", "metadata": {}, "outputs": [], "source": [ "\n", "## Defining the function\n", "\n", "def fetch_ref_data(access_token, format='json'):\n", "\n", " uri=\"v1.0/lng/hubs/cargo/reference-data/\"\n", " print(uri)\n", " \n", " content = do_api_get_query(\n", " uri=uri, access_token=access_token, format=format\n", " )\n", " \n", " if format == 'json':\n", " my_dict = content\n", " elif format == 'csv':\n", " # if there's no data to show, returns raw response (empty string) and \"No Data to Show\" message\n", " if len(content) == 0:\n", " my_dict = content\n", " print('No Data to Show')\n", " else:\n", " my_dict = content.decode('utf-8')\n", " my_dict = pd.read_csv(StringIO(my_dict)) # automatically converting into a Pandas DataFrame when choosing CSV format\n", " \n", " return my_dict\n", "\n", "ref_df = fetch_ref_data(access_token, format='csv')\n", "ref_df.head(5)" ] }, { "cell_type": "code", "execution_count": null, "id": "a6a14430", "metadata": {}, "outputs": [], "source": [ "# Fetch Live posts\n", "def fetch_active_posts(access_token, incoterm=None, region=None, format='json'):\n", "\n", " query_params = ''\n", "\n", " if incoterm is not None:\n", " query_params += '?incoterm={}'.format(incoterm)\n", " if region is not None:\n", " if len(query_params) >0:\n", " query_params += '®ion={}'.format(region)\n", " else:\n", " query_params += '?region={}'.format(region)\n", "\n", "\n", " uri=\"v1.0/lng/hubs/cargo/active/{}\".format(query_params)\n", " print(uri)\n", " \n", " content = do_api_get_query(\n", " uri=uri, access_token=access_token, format=format\n", " )\n", " \n", " if format == 'json':\n", " my_dict = content\n", " elif format == 'csv':\n", " # if there's no data to show, returns raw response (empty string) and \"No Data to Show\" message\n", " if len(content) == 0:\n", " my_dict = content\n", " print('No Data to Show')\n", " else:\n", " my_dict = content.decode('utf-8')\n", " my_dict = pd.read_csv(StringIO(my_dict)) # automatically converting into a Pandas DataFrame when choosing CSV format\n", " \n", " return my_dict\n", "\n", "# Fetch Historical Posts\n", "def fetch_inactive_posts(access_token, incoterm=None, region=None, format='json'):\n", "\n", " query_params = ''\n", "\n", " if incoterm is not None:\n", " query_params += '?incoterm={}'.format(incoterm)\n", " if region is not None:\n", " if len(query_params) >0:\n", " query_params += '®ion={}'.format(region)\n", " else:\n", " query_params += '?region={}'.format(region)\n", "\n", "\n", " uri=\"v1.0/lng/hubs/cargo/inactive/{}\".format(query_params)\n", " print(uri)\n", " \n", " content = do_api_get_query(\n", " uri=uri, access_token=access_token, format=format\n", " )\n", " \n", " if format == 'json':\n", " my_dict = content\n", " elif format == 'csv':\n", " # if there's no data to show, returns raw response (empty string) and \"No Data to Show\" message\n", " if len(content) == 0:\n", " my_dict = content\n", " print('No Data to Show')\n", " else:\n", " my_dict = content.decode('utf-8')\n", " my_dict = pd.read_csv(StringIO(my_dict)) # automatically converting into a Pandas DataFrame when choosing CSV format\n", " \n", " return my_dict" ] }, { "cell_type": "code", "execution_count": null, "id": "85d4678d", "metadata": {}, "outputs": [], "source": [ "# calling the live data\n", "usgc_livedf = fetch_active_posts(access_token, incoterm='fob', region='usg', format='csv')\n", "\n", "# calling the historical data\n", "usgc_histdf = fetch_inactive_posts(access_token, incoterm='fob', region='usg', format='csv')\n", "\n", "# Combining the dataframes to create a complete dataset\n", "total_df = pd.concat([usgc_livedf, usgc_histdf])\n", "\n", "# Converting dates to datetimes\n", "total_df['PostedAtUTC'] = pd.to_datetime(total_df['PostedAtUTC'])\n", "total_df['WindowStartDate'] = pd.to_datetime(total_df['WindowStartDate'])\n", "total_df['WindowEndDate'] = pd.to_datetime(total_df['WindowEndDate'])\n", "total_df['PostedDate'] = total_df['PostedAtUTC'].dt.strftime('%Y-%m-%d')\n", "\n", "# Creating a swaps-only dataframe\n", "swaps = total_df[total_df['InterestType'] == 'swap'].copy()\n", "\n", "swaps.head(5)" ] }, { "cell_type": "markdown", "id": "a6e8dd6e", "metadata": {}, "source": [ "# 3. Calling Spark30S Freight data\n", "\n", "Here we define the functions used to call the Spark30S Freight data from the Freight Contracts API. For more information on the Freight Contracts API, please visit the API website or download our Freight Contracts code sample:\n", "\n", "https://www.sparkcommodities.com/api/lng-freight/contracts.html" ] }, { "cell_type": "code", "execution_count": null, "id": "5ecbde02", "metadata": {}, "outputs": [], "source": [ "# Defining function to fetch historical freight data from API\n", "def fetch_historical_price_releases(access_token, ticker, limit, offset=None, vessel=None):\n", "\n", " query_params = \"?limit={}\".format(limit)\n", " if offset is not None:\n", " query_params += \"&offset={}\".format(offset)\n", " \n", " if vessel is not None:\n", " query_params += \"&vessel-type={}\".format(vessel)\n", "\n", " content = do_api_get_query(\n", " uri=\"/v1.0/contracts/{}/price-releases/{}\".format(ticker, query_params),\n", " access_token=access_token,\n", " )\n", "\n", " uri=\"/v1.0/contracts/{}/price-releases/{}\".format(ticker, query_params)\n", " print(uri)\n", "\n", " my_dict = content[\"data\"]\n", "\n", " return my_dict\n", "\n", "\n", "# Defining the function for storing and formatting the data into a Pandas DataFrame\n", "def store_and_format(dict_hist):\n", " stored_data = {\n", " \"ReleaseDate\": [],\n", " \"Ticker\": [],\n", " \"PeriodStart\": [],\n", " \"USDperday\": [],\n", " \"USDperdayMax\": [],\n", " \"USDperdayMin\": [],\n", " }\n", "\n", " for release in dict_hist:\n", "\n", " data_points = release[\"data\"][0][\"dataPoints\"]\n", "\n", " for data_point in data_points:\n", " stored_data['Ticker'].append(release[\"contractId\"])\n", " stored_data['ReleaseDate'].append(release[\"releaseDate\"])\n", "\n", " period_start_at = data_point[\"deliveryPeriod\"][\"startAt\"]\n", " stored_data['PeriodStart'].append(period_start_at)\n", "\n", " stored_data['USDperday'].append(data_point[\"derivedPrices\"][\"usdPerDay\"][\"spark\"])\n", " stored_data['USDperdayMin'].append(data_point[\"derivedPrices\"][\"usdPerDay\"][\"sparkMin\"])\n", " stored_data['USDperdayMax'].append(data_point[\"derivedPrices\"][\"usdPerDay\"][\"sparkMax\"])\n", " \n", " historical_df = pd.DataFrame(stored_data)\n", " \n", " historical_df[\"USDperday\"] = pd.to_numeric(historical_df[\"USDperday\"])\n", " historical_df[\"USDperdayMax\"] = pd.to_numeric(historical_df[\"USDperdayMax\"])\n", " historical_df[\"USDperdayMin\"] = pd.to_numeric(historical_df[\"USDperdayMin\"])\n", "\n", " historical_df[\"ReleaseDate\"] = pd.to_datetime(historical_df[\"ReleaseDate\"])\n", " \n", " return historical_df" ] }, { "cell_type": "code", "execution_count": null, "id": "f1b2ac0c", "metadata": {}, "outputs": [], "source": [ "# determining approximate length of freight data needed based on swaps date range\n", "freight_length = (swaps['PostedAtUTC'].max() - swaps['PostedAtUTC'].min()).days" ] }, { "cell_type": "code", "execution_count": null, "id": "b9e09634", "metadata": {}, "outputs": [], "source": [ "### Calling Spark30S data using above functions\n", "spark30s_raw = fetch_historical_price_releases(access_token, 'spark30s', limit=freight_length, offset=0)\n", "spark30s = store_and_format(spark30s_raw)\n", "\n", "spark30s.head(5)" ] }, { "cell_type": "markdown", "id": "4af1c14b", "metadata": {}, "source": [ "# 4. Computing Swaps \"Momentum\" metric\n", "\n", "Here we define a function to compute the \"momentum\" metric for the swaps data. This iterates through all swap posts to classify them as either \"Request for Later Loading Slot\" or \"Request for Earlier Loading Slot\".\n", "\n", "The function includes several options to filter the data & which posts to include within this metric" ] }, { "cell_type": "code", "execution_count": null, "id": "7653442d", "metadata": {}, "outputs": [], "source": [ "\n", "def compute_swap_momentum(posts_df, spot_window=False, window_days=30,\n", " normalise=False, min_diff=0, max_diff=None):\n", " f\"\"\"\n", " Parameters\n", " ----------\n", " spot_window : bool, default False\n", " True -> a post counts only if EITHER its offered or requested date sits within the (T -> T+\"window_days\") window, \n", " where T = the post's PostedAt date).\n", " window_days : int, default 30\n", " The length of window to apply for the \"spot_window\" param (T+30 by default to match Spark30S spec; T = the post's PostedAt date).\n", " normalise : bool, default False\n", " Normalise for the amount of posts per month\n", " min_diff : int, default 0\n", " Only count posts whose |requested - offered| shift is >= min_diff days. 0 = no filtering.\n", " max_diff : int or None, default None\n", " Only count posts whose |requested - offered| shift is <= min_diff days. 0 = no filtering.\n", "\n", " Returns\n", " -------\n", " moments_df = momentum metric per release date\n", " months_df = momentum metric summed over the month\n", " \"\"\"\n", "\n", " # Creating a copy of the dataframe, and making sure it only has Swaps\n", " df = posts_df.copy()\n", " df = df[df['InterestType'] == 'swap']\n", "\n", " df['PostedAtUTC'] = pd.to_datetime(df['PostedAtUTC'])\n", " df['WindowStartDate'] = pd.to_datetime(df['WindowStartDate'])\n", " df['PostedDate'] = df['PostedAtUTC'].dt.strftime('%Y-%m-%d')\n", "\n", " # Initiating lists\n", " datetimes, later, earlier = [], [], []\n", "\n", " # For each post date\n", " for r in df['PostedDate'].unique():\n", " rdf = df[df['PostedDate'] == r] # filter dataframe per post date\n", " l = e = 0 # initiate later and earlier counters\n", "\n", " t_start = pd.to_datetime(r)\n", " t_end = t_start + pd.Timedelta(days=window_days)\n", "\n", " for pid in rdf['PostID'].unique(): # Iterating through every post\n", " tdf = rdf[rdf['PostID'] == pid]\n", " offered = tdf[tdf['OrderType'] == 'offered']\n", " requested = tdf[tdf['OrderType'] == 'requested']\n", " if offered.empty or requested.empty: # Check if empty\n", " continue\n", "\n", " offer = offered['WindowStartDate'].min() # earliest offered\n", " sell = requested['WindowStartDate'].min() # earliest requested\n", "\n", " if spot_window: # Apply spot_window filtering if specified\n", " offer_in = t_start <= offer <= t_end \n", " sell_in = t_start <= sell <= t_end\n", " if not (offer_in or sell_in):\n", " continue\n", " \n", " shift = abs((sell - offer).days) # Apply min_diff or max_diff if specified\n", " if shift < min_diff:\n", " continue\n", " if max_diff is not None and shift > max_diff:\n", " continue\n", "\n", " if offer < sell: # Counting each post as \"earlier\" or \"later\" loading\n", " l += 1\n", " elif sell < offer:\n", " e += 1\n", "\n", " datetimes.append(rdf['PostedAtUTC'].iloc[0])\n", " later.append(l)\n", " earlier.append(e)\n", "\n", " def _diff(later_s, earlier_s): # Creating an internal function to calculate metric\n", " net = later_s - earlier_s\n", " if normalise: # Applying normalisation if specified\n", " total = later_s + earlier_s\n", " return np.where(total > 0, net / total, np.nan)\n", " return net\n", "\n", " moments_df = pd.DataFrame({ # Creating dataframe with per-post_date metrics\n", " 'Date': pd.to_datetime(datetimes),\n", " 'Later': later,\n", " 'Earlier': earlier,\n", " })\n", " moments_df['Diff'] = _diff(moments_df['Later'], moments_df['Earlier'])\n", " moments_df = moments_df.sort_values('Date', ascending=False).reset_index(drop=True)\n", " moments_df['Month'] = moments_df['Date'].dt.strftime('%b%y')\n", " moments_df = moments_df[['Date', 'Month', 'Earlier', 'Later', 'Diff']]\n", "\n", " months_df = ( # Summing by month\n", " moments_df\n", " .groupby('Month', as_index=False)[['Earlier', 'Later']]\n", " .sum()\n", " )\n", " months_df['Diff'] = _diff(months_df['Later'], months_df['Earlier'])\n", " months_df['MonthDatetime'] = pd.to_datetime(months_df['Month'], format='%b%y')\n", " months_df = months_df.sort_values('MonthDatetime').reset_index(drop=True)\n", " months_df = months_df[['Month', 'MonthDatetime', 'Earlier', 'Later', 'Diff']]\n", "\n", " return moments_df, months_df" ] }, { "cell_type": "code", "execution_count": null, "id": "04aa1429", "metadata": {}, "outputs": [], "source": [ "# Applying function to swaps dataframe \n", "moments_df, months_df = compute_swap_momentum(swaps)\n", "\n", "# Preview per-post_date data \n", "moments_df.head(3)" ] }, { "cell_type": "code", "execution_count": null, "id": "a23f1003", "metadata": {}, "outputs": [], "source": [ "# Preview monthly data\n", "months_df" ] }, { "cell_type": "markdown", "id": "eb07c9b2", "metadata": {}, "source": [ "# 5. Plotting\n", "\n", "Here we plot the Spark30S Freight rates vs the Monthly USGC Swaps \"Momentum\". \n", "\n", "__N.B.__ We use the \"months_df\" dataframe defined in the cells above, where no additional filter parameters were specified when calling the \"compute_swap_momentum\" function." ] }, { "cell_type": "code", "execution_count": null, "id": "3d35ee57", "metadata": {}, "outputs": [], "source": [ "import matplotlib.pyplot as plt\n", "import seaborn as sns\n", "\n", "sns.set_style('whitegrid')\n", "\n", "fig1, ax1 = plt.subplots(figsize=(16,7))\n", "\n", "# creating two separate dataframe views to plot \"Net Later\" and \"Net Earlier\" posts in separate colours \n", "pos_df = months_df[months_df['Diff'] >0]\n", "neg_df = months_df[months_df['Diff'] <0]\n", "\n", "# Plotting thicker x-axis line\n", "ax1.hlines(0, months_df['MonthDatetime'].iloc[1] - pd.Timedelta(days=60), months_df['MonthDatetime'].iloc[-1] + pd.Timedelta(days=60), color='grey')\n", "\n", "# Plotting Swaps momentum metric bar charts\n", "ax1.bar(pos_df['MonthDatetime'] + pd.Timedelta(days=15), pos_df['Diff'], color='mediumseagreen', width=27, align='center', label='Later')\n", "ax1.bar(neg_df['MonthDatetime'] + pd.Timedelta(days=15), neg_df['Diff'], color='firebrick', width=27, align='center', label='Earlier')\n", "plt.legend(loc='lower right')\n", "\n", "# Limiting date range to match swaps data\n", "ax1.set_xlim(months_df['MonthDatetime'].iloc[0] - pd.Timedelta(days=10), months_df['MonthDatetime'].iloc[-1] + pd.Timedelta(days=40))\n", "ax1.set_ylim(-15, 10)\n", "\n", "# Creating twin-axis to plot Spark30S freight rates\n", "ax12 = ax1.twinx()\n", "\n", "# Plotting Spark30S\n", "ax12.plot(spark30s['ReleaseDate'], spark30s['USDperday'], linewidth=2, color='mediumblue', alpha=0.8, zorder=0, label='Spark30S (Atlantic)')\n", "ax12.scatter(spark30s['ReleaseDate'].iloc[0], spark30s['USDperday'].iloc[0], color='mediumblue', marker='o', alpha=0.8, s=80)\n", "\n", "ax12.grid(False)\n", "plt.legend(loc='upper right')" ] } ], "metadata": { "kernelspec": { "display_name": "base", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.13.9" } }, "nbformat": 4, "nbformat_minor": 5 }