`turtle_quant_1.data_processing.adapters.gcs_storage_adapter`¶

turtle_quant_1.data_processing.adapters.gcs_storage_adapter ¶

Google Cloud Storage implementation for data storage.

GCSDataStorageAdapter ¶

GCSDataStorageAdapter(
    bucket_name: Optional[str] = None,
    project_id: Optional[str] = None,
)

Bases: BaseDataStorageAdapter

Google Cloud Storage implementation for data storage.

Initialize the GCS data storage.

Parameters:

Name	Type	Description	Default
`bucket_name`	`Optional[str]`	Name of the GCS bucket. If None, reads from GCS_BUCKET_NAME env var.	`None`
`project_id`	`Optional[str]`	GCP project ID. If None, reads from GCP_PROJECT_ID env var.	`None`

Source code in src/turtle_quant_1/data_processing/adapters/gcs_storage_adapter.py

def __init__(
    self,
    bucket_name: Optional[str] = None,
    project_id: Optional[str] = None,
):
    """Initialize the GCS data storage.

    Args:
        bucket_name: Name of the GCS bucket. If None, reads from GCS_BUCKET_NAME env var.
        project_id: GCP project ID. If None, reads from GCP_PROJECT_ID env var.
    """
    self.project_id = project_id or GCLOUD_PROJECT_ID
    self.bucket_name = bucket_name or GCLOUD_STB_DATA_NAME

    self.client = storage.Client(project=self.project_id)
    self.bucket = self.client.bucket(self.bucket_name)

delete_ohlcv_data ¶

delete_ohlcv_data(
    symbol: str, freq: CandleUnit = CANDLE_UNIT
) -> None

Delete data from GCS.

Parameters:

Name	Type	Description	Default
`symbol`	`str`	The symbol to delete data for.	required
`freq`	`CandleUnit`	The frequency of the data to delete.	`CANDLE_UNIT`

Source code in src/turtle_quant_1/data_processing/adapters/gcs_storage_adapter.py

def delete_ohlcv_data(self, symbol: str, freq: CandleUnit = CANDLE_UNIT) -> None:
    """Delete data from GCS.

    Args:
        symbol: The symbol to delete data for.
        freq: The frequency of the data to delete.
    """
    blob_name = self._get_blob_name(symbol, freq)
    blob = self.bucket.blob(blob_name)

    if blob.exists():
        blob.delete()

load_ohlcv_data ¶

load_ohlcv_data(
    symbol: str,
    start_date: Optional[datetime] = None,
    end_date: Optional[datetime] = None,
    freq: CandleUnit = CANDLE_UNIT,
) -> DataFrame

Load OHLCV data from GCS.

Parameters:

Name	Type	Description	Default
`symbol`	`str`	The symbol to load data for.	required
`start_date`	`Optional[datetime]`	Start date for the data. If None, returns all data.	`None`
`end_date`	`Optional[datetime]`	End date for the data. If None, returns all data.	`None`
`freq`	`CandleUnit`	The frequency of the data.	`CANDLE_UNIT`

Returns:

Type	Description
`DataFrame`	DataFrame with columns: datetime, open, high, low, close, volume.

Source code in src/turtle_quant_1/data_processing/adapters/gcs_storage_adapter.py

def load_ohlcv_data(
    self,
    symbol: str,
    start_date: Optional[datetime] = None,
    end_date: Optional[datetime] = None,
    freq: CandleUnit = CANDLE_UNIT,
) -> pd.DataFrame:
    """Load OHLCV data from GCS.

    Args:
        symbol: The symbol to load data for.
        start_date: Start date for the data. If None, returns all data.
        end_date: End date for the data. If None, returns all data.
        freq: The frequency of the data.

    Returns:
        DataFrame with columns: datetime, open, high, low, close, volume.
    """
    blob_name = self._get_blob_name(symbol, freq)
    blob = self.bucket.blob(blob_name)

    # Check if blob exists
    if not blob.exists():
        return pd.DataFrame(
            columns=["datetime", "Open", "High", "Low", "Close", "Volume"]
        )

    # Download and read parquet file
    with io.BytesIO() as buffer:
        blob.download_to_file(buffer)
        buffer.seek(0)
        df = pd.read_parquet(buffer)

    # Filter by date range if provided
    if start_date is not None and end_date is not None and not df.empty:
        mask = (df["datetime"] >= start_date) & (df["datetime"] <= end_date)
        return df[mask]

    return df

save_ohlcv_data ¶

save_ohlcv_data(
    symbol: str,
    data: DataFrame,
    freq: CandleUnit = CANDLE_UNIT,
) -> None

Save OHLCV data to GCS.

Parameters:

Name	Type	Description	Default
`symbol`	`str`	The symbol the data belongs to.	required
`data`	`DataFrame`	DataFrame with OHLCV data.	required

Source code in src/turtle_quant_1/data_processing/adapters/gcs_storage_adapter.py

def save_ohlcv_data(
    self,
    symbol: str,
    data: pd.DataFrame,
    freq: CandleUnit = CANDLE_UNIT,
) -> None:
    """Save OHLCV data to GCS.

    Args:
        symbol: The symbol the data belongs to.
        data: DataFrame with OHLCV data.
    """
    blob_name = self._get_blob_name(symbol, freq)
    blob = self.bucket.blob(blob_name)

    # Convert DataFrame to parquet and upload
    with io.BytesIO() as buffer:
        data.to_parquet(buffer)
        buffer.seek(0)
        blob.upload_from_file(buffer, content_type="application/octet-stream")

turtle_quant_1.data_processing.adapters.gcs_storage_adapter¶