Create loggers per module

Author: jspahrsummers

blotter/__main__.py

@@ -16,6 +16,8 @@
 from blotter.server import Servicer
 from blotter.streaming import StreamingManager
 
+logger = logging.getLogger(__package__)
+
 parser = ArgumentParser(
     prog="blotter",
     description="Microservice to connect to Interactive Brokers and stream market data into Google BigQuery",
@@ -98,7 +100,7 @@ def main() -> None:
 
     ib = ib_insync.IB()
 
-    logging.info(f"Connecting to IB on {args.tws_host}:{args.tws_port}")
+    logger.info(f"Connecting to IB on {args.tws_host}:{args.tws_port}")
     ib.connect(
         host=args.tws_host,
         port=args.tws_port,
@@ -115,9 +117,9 @@ def handle_ib_thread_error(error: Exception) -> None:
             try:
                 raise error
             except IBWarning:
-                logging.warning(f"Warning from IB: {error}")
+                logger.warning(f"Warning from IB: {error}")
             except ConnectionError:
-                logging.exception(f"Connection error from IB:")
+                logger.exception(f"Connection error from IB:")
 
     thread = IBThread(ib, error_handler=handle_ib_thread_error)
     port = args.port or random.randint(49152, 65535)
@@ -131,7 +133,7 @@ def handle_ib_thread_error(error: Exception) -> None:
     if args.resume:
         servicer.resume_streaming()
 
-    logging.info(f"Server listening on port {port}")
+    logger.info(f"Server listening on port {port}")
     thread.run_forever()
 
 

blotter/backfill.py

@@ -1,5 +1,5 @@
-import logging
 from datetime import datetime
+from logging import getLogger
 from typing import Tuple
 
 import ib_insync
@@ -10,11 +10,9 @@
 from blotter.blotter_pb2 import ContractSpecifier
 from blotter.error_handling import ErrorHandlerConfiguration
 from blotter.ib_helpers import DataError, qualify_contract_specifier
-from blotter.upload import (
-    BarsTableColumn,
-    table_name_for_contract,
-    upload_dataframe,
-)
+from blotter.upload import BarsTableColumn, table_name_for_contract, upload_dataframe
+
+logger = getLogger(__name__)
 
 
 async def backfill_bars(
@@ -50,7 +48,7 @@ async def backfill_bars(
 
     earliest_date = barList[0].date
 
-    logging.info(f"Loaded {len(barList)} historical bars for {con}")
+    logger.info(f"Loaded {len(barList)} historical bars for {con}")
     df = ib_insync.util.df(barList)
 
     # See fields on BarData.
@@ -69,7 +67,7 @@ async def backfill_bars(
 
     df[BarsTableColumn.BAR_SOURCE.value] = barList.whatToShow
 
-    logging.debug(df)
+    logger.debug(df)
     job = upload_dataframe(table_name_for_contract(con), df, error_handler)
 
     return (earliest_date, job)

blotter/ib_helpers.py

@@ -1,6 +1,5 @@
 import asyncio
 import concurrent.futures
-import logging
 import math
 from dataclasses import dataclass
 from decimal import Decimal

blotter/options.py

@@ -1,5 +1,5 @@
-import logging
 from datetime import datetime
+from logging import getLogger
 from typing import Any, Dict, Iterable, List, Union, cast
 
 import ib_insync
@@ -11,6 +11,8 @@
 from blotter.ib_helpers import qualify_contract_specifier, sanitize_price
 from blotter.upload import TickersTableColumn, table_name_for_contract, upload_dataframe
 
+logger = getLogger(__name__)
+
 
 async def _look_up_options(
     ib_client: ib_insync.IB, underlying: ib_insync.Contract,
@@ -25,7 +27,7 @@ async def _look_up_options(
         underlyingConId=underlying.conId,
     )
 
-    logging.info(f"Loaded {len(option_chains)} option chains for {underlying}")
+    logger.info(f"Loaded {len(option_chains)} option chains for {underlying}")
 
     option_contracts = (
         ib_insync.Option(
@@ -46,7 +48,7 @@ async def _look_up_options(
 
     qualified_contracts = await ib_client.qualifyContractsAsync(*option_contracts)
 
-    logging.info(
+    logger.info(
         f"Qualified {len(qualified_contracts)} options contracts for {underlying}"
     )
 
@@ -63,7 +65,7 @@ async def _load_tickers_into_dataframe(
     """
 
     tickers = await ib_client.reqTickersAsync(*contracts, regulatorySnapshot=False)
-    logging.info(f"Fetched {len(tickers)} tickers")
+    logger.info(f"Fetched {len(tickers)} tickers")
 
     def _ticker_dict(t: Any,) -> Dict[str, Union[str, datetime, float, None]]:  # FIXME
         price_is_negative = t.close < 0
@@ -121,7 +123,7 @@ def _ticker_dict(t: Any,) -> Dict[str, Union[str, datetime, float, None]]:  # FI
         TickersTableColumn.TIMESTAMP.value
     ].dt.round("ms")
 
-    logging.debug(f"Tickers DataFrame: {df}")
+    logger.debug(f"Tickers DataFrame: {df}")
     return df
 
 

blotter/server.py

@@ -1,6 +1,6 @@
 import asyncio
 import concurrent.futures
-import logging
+from logging import getLogger
 from datetime import datetime, timedelta, timezone
 from typing import Awaitable, Callable, Iterator, Optional, Tuple, TypeVar
 
@@ -14,6 +14,8 @@
 from blotter.streaming import StreamingID, StreamingManager
 from google.cloud import bigquery
 
+logger = getLogger(__name__)
+
 _T = TypeVar("_T")
 
 
@@ -64,7 +66,7 @@ def resume_streaming(self) -> None:
         """
 
         streaming_ids = list(self._streaming_manager.resume_streaming(self._ib_thread))
-        logging.info(f"Resumed streaming IDs {streaming_ids}")
+        logger.info(f"Resumed streaming IDs {streaming_ids}")
 
     def _run_in_ib_thread(
         self, fn: Callable[[ib_insync.IB], Awaitable[_T]]
@@ -87,7 +89,7 @@ def LoadHistoricalData(
         request: blotter_pb2.LoadHistoricalDataRequest,
         context: grpc.ServicerContext,
     ) -> Iterator[blotter_pb2.LoadHistoricalDataResponse]:
-        logging.info(f"LoadHistoricalData: {request}")
+        logger.info(f"LoadHistoricalData: {request}")
 
         td = request_helpers.duration_timedelta_atleast(request.duration)
         end_date = datetime.fromtimestamp(request.endTimestampUTC, tz=timezone.utc)
@@ -96,7 +98,7 @@ def LoadHistoricalData(
             duration = request_helpers.duration_str(request.duration)
             start_date = end_date - timedelta(seconds=1)
         else:
-            logging.debug(f"Splitting requested duration {td}")
+            logger.debug(f"Splitting requested duration {td}")
             duration = request_helpers.duration_str(
                 blotter_pb2.Duration(count=10, unit=blotter_pb2.Duration.TimeUnit.DAYS)
             )
@@ -108,7 +110,7 @@ async def _backfill(
         ) -> Tuple[datetime, bigquery.LoadJob]:
             nonlocal end_date
 
-            logging.info(
+            logger.info(
                 f"Backfilling {duration} from {end_date} of {request.contractSpecifier}"
             )
 
@@ -126,15 +128,15 @@ async def _backfill(
         while end_date > start_date:
             (end_date, job) = self._run_in_ib_thread(_backfill).result()
 
-            logging.info(f"BigQuery backfill job launched: {job.job_id}")
+            logger.info(f"BigQuery backfill job launched: {job.job_id}")
             yield blotter_pb2.LoadHistoricalDataResponse(backfillJobID=job.job_id)
 
     def StartRealTimeData(
         self,
         request: blotter_pb2.StartRealTimeDataRequest,
         context: grpc.ServicerContext,
     ) -> blotter_pb2.StartRealTimeDataResponse:
-        logging.info(f"StartRealTimeData: {request}")
+        logger.info(f"StartRealTimeData: {request}")
 
         async def _start_stream(ib_client: ib_insync.IB) -> StreamingID:
             return await self._streaming_manager.start_stream(
@@ -145,7 +147,7 @@ async def _start_stream(ib_client: ib_insync.IB) -> StreamingID:
             )
 
         streaming_id = self._run_in_ib_thread(_start_stream).result()
-        logging.debug(f"Real-time bars streaming ID: {streaming_id}")
+        logger.debug(f"Real-time bars streaming ID: {streaming_id}")
 
         return blotter_pb2.StartRealTimeDataResponse(requestID=streaming_id)
 
@@ -154,7 +156,7 @@ def CancelRealTimeData(
         request: blotter_pb2.CancelRealTimeDataRequest,
         context: grpc.ServicerContext,
     ) -> blotter_pb2.CancelRealTimeDataResponse:
-        logging.info(f"CancelRealTimeData: {request}")
+        logger.info(f"CancelRealTimeData: {request}")
 
         async def _cancel_stream(ib_client: ib_insync.IB) -> None:
             await self._streaming_manager.cancel_stream(
@@ -167,23 +169,23 @@ async def _cancel_stream(ib_client: ib_insync.IB) -> None:
     def HealthCheck(
         self, request: blotter_pb2.HealthCheckRequest, context: grpc.ServicerContext,
     ) -> blotter_pb2.HealthCheckResponse:
-        logging.info(f"HealthCheck: {request}")
+        logger.info(f"HealthCheck: {request}")
         return blotter_pb2.HealthCheckResponse()
 
     def SnapshotOptionChain(
         self,
         request: blotter_pb2.SnapshotOptionChainRequest,
         context: grpc.ServicerContext,
     ) -> blotter_pb2.SnapshotOptionChainResponse:
-        logging.info(f"SnapshotOptionChain: {request}")
+        logger.info(f"SnapshotOptionChain: {request}")
 
         async def _snapshot(ib_client: ib_insync.IB) -> bigquery.LoadJob:
             return await snapshot_options(
                 ib_client, request.contractSpecifier, self._error_handler
             )
 
         job = self._run_in_ib_thread(_snapshot).result()
-        logging.info(f"BigQuery import job launched: {job.job_id}")
+        logger.info(f"BigQuery import job launched: {job.job_id}")
 
         return blotter_pb2.SnapshotOptionChainResponse(importJobID=job.job_id)