refactor(market-data-ingestion): change polygon.io to massive.com

Change stocks data HTTP client from polygon.io to massive.com following its rebranding.

Author: guilhem-sante

data-ingestion/README.md

@@ -10,7 +10,7 @@ Data pipeline to ingest stock market datasets into a Time Series Database.
 2. Defines the required environment variables
 
 ```sh
-export POLYGON_API_KEY=...
+export MASSIVE_API_KEY=...
 
 export POSTGRES_USER="finwar"
 export POSTGRES_PASSWORD="password"
@@ -21,7 +21,7 @@ export POSTGRES_DB="finwar"
 
 
 > [!TIP]
-> You can get a Polygon API key for free by creating an account on [polygon.io](https://polygon.io/).
+> You can get a Massive API key for free by creating an account on [massive.com](https://massive.com/).
 
 ### Installing dependencies
 
@@ -57,4 +57,4 @@ uv run dg launch --assets download_stocks_intraday_history,load_stocks_intraday_
 *here the pipeline is launched to collect the intraday history data from Apple Inc. on the last week of september.*
 
 > [!WARNING]  
-> The Polygon API free account only goes up to 5 req/min so beware in the number of parition you run.
+> The Massive API free account only goes up to 5 req/min so beware in the number of parition you run.

data-ingestion/docker/Dockerfile

@@ -2,7 +2,7 @@ FROM docker.io/library/python:3.13.7-slim AS builder
 
 ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy UV_PYTHON_DOWNLOADS=0
 
-COPY --from=ghcr.io/astral-sh/uv:0.8.22 /uv /uvx /bin/
+COPY --from=ghcr.io/astral-sh/uv:0.9.6 /uv /uvx /bin/
 
 WORKDIR /app
 
@@ -14,7 +14,12 @@ COPY . /app
 
 RUN uv sync --no-dev --locked
 
-FROM docker.io/library/python:3.13.7-slim AS runtime
+FROM docker.io/library/python:3.13.7-slim AS runner
+
+LABEL org.opencontainers.image.authors="guilhem.sante@kelps.org"
+LABEL org.opencontainers.image.title="FinWar Data Ingestion User Code"
+LABEL org.opencontainers.image.description="Data pipeline to ingest stock market datasets into a Time Series Database."
+LABEL org.opencontainers.image.licenses="MIT"
 
 RUN groupadd --system --gid 999 appuser \
  && useradd --system --gid 999 --uid 999 --create-home appuser

data-ingestion/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "finwar_data_ingestion"
-version = "0.1.0"
+version = "0.2.0"
 description = "Data pipeline to ingest stock market datasets into a Time Series Database"
 authors = [
     { name = "Coding Kelps" }
@@ -10,8 +10,8 @@ requires-python = "==3.13.*"
 dependencies = [
     "dagster==1.11.10",
     "dagster-docker>=0.27.10",
+    "massive>=2.0.1",
     "pandas>=2.3.2",
-    "polygon>=1.2.8",
     "psycopg[binary]>=3.2.10",
     "python-dateutil>=2.9.0.post0",
 ]

data-ingestion/src/finwar_data_ingestion/components/stocks_intraday_history.py

@@ -4,16 +4,16 @@
 from datetime import datetime
 from typing import Literal, List
 
-from finwar_data_ingestion.resources import PostgreSQLResource, PolygonResource
+from finwar_data_ingestion.resources import PostgreSQLResource, MassiveResource
 
 
-MAX_POLYGON_REQ_RETRY = 3
+MAX_MASSIVE_REQ_RETRY = 3
 
 
 class StocksIntradayHistoryComponent(dg.Component, dg.Model, dg.Resolvable):
 	"""Represents historical stock quotation for one or more ticker symbols.
 
-	The data is retrieved from the [Polygon.io](https://polygon.io) API and returned as a
+	The data is retrieved from the [Massive](https://massive.com) API and returned as a
 	:class:`pandas.DataFrame`. The data also contains the trading volume.
 	It is then loaded as a TimescaleDB table.
 
@@ -25,7 +25,7 @@ class StocksIntradayHistoryComponent(dg.Component, dg.Model, dg.Resolvable):
 	    time_partitioning (Literal["daily", "weekly", "monthly"]): The partitioning frequency
 	        for the data retrieval. The partitioning unit should be shorter or equal to the
 	        requested time window. Default is ``"monthly"``.
-	    polygon_api_key (str): API key for authenticating with the Polygon.io API.
+	    massive_api_key (str): API key for authenticating with the Massive API.
 	    precision (Literal["second", "minute", "day", "week", "month", "quarter", "year"]):
 	        The granularity of the bars. Must be finer than or equal to the requested
 	        time window. Default is ``"day"``.
@@ -42,7 +42,7 @@ class StocksIntradayHistoryComponent(dg.Component, dg.Model, dg.Resolvable):
 	start_date: str
 	end_date: str | None = None
 	time_partitioning: Literal['daily', 'weekly', 'monthly'] = 'monthly'
-	polygon_api_key: str
+	massive_api_key: str
 	precision: Literal['second', 'minute', 'day', 'week', 'month', 'quarter', 'year'] = 'day'
 	adjusted: bool = False
 	table_name: str = 'stocks_history'
@@ -57,7 +57,7 @@ def get_spec(cls) -> dg.ComponentTypeSpec:
 		return dg.ComponentTypeSpec(
 			description=cls.__doc__,
 			owners=['contact@kelps.org', 'guilhem.sante@kelps.org'],
-			tags=['stocks', 'intraday', 'polygon.io', 'timescaledb'],
+			tags=['stocks', 'intraday', 'massive.com', 'timescaledb'],
 		)
 
 	def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
@@ -84,57 +84,57 @@ def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
 			group_name='ingestion',
 			code_version='0.3.0',
 			description="""
-            Extract the stocks intraday value from the Polygon API.
+            Extract the stocks intraday value from the Massive API.
           """,
 			partitions_def=partitions_matrix,
 		)
 		def download_stocks_intraday_history(
 			context: AssetExecutionContext,
-			polygon: PolygonResource,
+			massive: MassiveResource,
 		) -> dg.MaterializeResult:
 			partition_keys = context.partition_key.keys_by_dimension
 			time_window = time_partitioning.time_window_for_partition_key(partition_keys['time'])
 			symbol = partition_keys['symbol']
 
-			with polygon.get_client() as client:
-				for attempt in range(0, MAX_POLYGON_REQ_RETRY):
-					try:
-						res = client.get_aggregate_bars(
-							symbol=symbol,
-							multiplier=1,
-							adjusted=self.adjusted,
-							timespan=self.precision,
-							from_date=time_window.start,
-							to_date=time_window.end,
-							limit=50000,  # API max limit
+			client = massive.get_client()
+
+			for attempt in range(0, MAX_MASSIVE_REQ_RETRY):
+				try:
+					aggs = []
+					for a in client.list_aggs(
+						ticker=symbol,
+						multiplier=1,
+						adjusted=self.adjusted,
+						timespan=self.precision,
+						from_=time_window.start,
+						to=time_window.end,
+						limit=50000,  # API max limit
+					):
+						aggs.append(a)
+
+				except Exception as e:
+					if attempt == MAX_MASSIVE_REQ_RETRY:
+						raise dg.Failure(
+							description=f'failed to get intraday stock data from Massive API: {e}'
+						)
+					else:
+						context.log.warning(
+							f'Massive API request failed (attempt n°{attempt + 1}/{MAX_MASSIVE_REQ_RETRY}): {e}',
 						)
-
-						if res['status'] == 'ERROR' or res['results'] is None:
-							raise Exception(res['error'])
-						break
-					except Exception as e:
-						if attempt == MAX_POLYGON_REQ_RETRY:
-							raise dg.Failure(
-								description=f'failed to get intraday stock data from Polygon API: {e}'
-							)
-						else:
-							context.log.warning(
-								f'polygon.io API request failed (attempt n°{attempt + 1}/{MAX_POLYGON_REQ_RETRY}): {e}',
-							)
 
 			return dg.MaterializeResult(
 				value=pd.DataFrame(
 					[
 						{
-							'timestamp': datetime.fromtimestamp(bar['t'] / 1000),
+							'timestamp': datetime.fromtimestamp(bar.timestamp / 1000),
 							'symbol': symbol,
-							'open': bar['o'],
-							'high': bar['h'],
-							'low': bar['l'],
-							'close': bar['c'],
-							'volume': bar['v'],
+							'open': bar.open,
+							'high': bar.high,
+							'low': bar.low,
+							'close': bar.close,
+							'volume': bar.volume,
 						}
-						for bar in res['results']
+						for bar in aggs
 					]
 				),
 				metadata={
@@ -222,8 +222,8 @@ def load_stocks_intraday_history(
 				load_stocks_intraday_history,
 			],
 			resources={
-				'polygon': PolygonResource(
-					api_key=self.polygon_api_key,
+				'massive': MassiveResource(
+					api_key=self.massive_api_key,
 				),
 				'postgresql': PostgreSQLResource(
 					username=self.timescaledb_username,

data-ingestion/src/finwar_data_ingestion/defs/stocks_intraday_history/defs.yaml

@@ -57,7 +57,7 @@ attributes:
     - CAT
     - AMGN
     - NEE
-  polygon_api_key: "{{ env.POLYGON_API_KEY }}"
+  massive_api_key: "{{ env.MASSIVE_API_KEY }}"
   timescaledb_username: "{{ env.POSTGRES_USER }}"
   timescaledb_password: "{{ env.POSTGRES_PASSWORD }}"
   timescaledb_host: "{{ env.POSTGRES_HOST }}"
@@ -66,7 +66,7 @@ attributes:
 
 requirements:
   env:
-    - POLYGON_API_KEY
+    - MASSIVE_API_KEY
     - POSTGRES_USER
     - POSTGRES_PASSWORD
     - POSTGRES_HOST

data-ingestion/src/finwar_data_ingestion/resources/__init__.py

@@ -1,2 +1,2 @@
 from .postgresql import PostgreSQLResource as PostgreSQLResource
-from .polygon import PolygonResource as PolygonResource
+from .massive import MassiveResource as MassiveResource

data-ingestion/src/finwar_data_ingestion/resources/massive.py

@@ -0,0 +1,17 @@
+from dagster import ConfigurableResource
+from pydantic import Field
+
+from massive import RESTClient
+
+
+class MassiveResource(ConfigurableResource):
+	"""
+	A Dagster resource wrapper over the massive.com REST client.
+	"""
+
+	api_key: str = Field(description='The required API key to authenticate.')
+
+	def get_client(self) -> RESTClient:
+		return RESTClient(
+			api_key=self.api_key,
+		)