Add background task progress logging

App/routers/bonds/routes.py

@@ -1,3 +1,4 @@
+import logging
 from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
 from tortoise.contrib.pydantic.creator import pydantic_queryset_creator
 from tortoise.transactions import in_transaction
@@ -12,6 +13,7 @@ from datetime import date
 
 
 router = APIRouter(prefix="/bonds", tags=["Bonds"])
+logger = logging.getLogger(__name__)
 
 
 def _bond_time_left(maturity_date):
@@ -104,6 +106,7 @@ async def run_bond_import_task(task_id: int):
     updated_count = 0
     failed_count = 0
     processed_isins = set()
+    logger.info("[bond-import:%s] background import started", task_id)
 
     try:
         async for bond_data in scraper.scrape_all_bond_data():
@@ -112,7 +115,11 @@ async def run_bond_import_task(task_id: int):
                 continue
             
             if bond_data.isin and bond_data.isin in processed_isins:
-                print(f"Skipping duplicate ISIN in current scrape: {bond_data.isin}")
+                logger.info(
+                    "[bond-import:%s] duplicate isin=%s skipped",
+                    task_id,
+                    bond_data.isin,
+                )
                 continue
 
             async with in_transaction(): # Ensure atomicity for each bond
@@ -133,18 +140,33 @@ async def run_bond_import_task(task_id: int):
                     if existing_bond:
                         await Bond.filter(id=existing_bond.id).update(**bond_data.dict(exclude_unset=True))
                         updated_count += 1
-                        print(f"Updated bond: ISIN {bond_data.isin}, AuNo {bond_data.auction_number}")
+                        action = "updated"
                     else:
                         await Bond.create(**bond_data.dict())
                         created_count += 1
-                        print(f"Created bond: ISIN {bond_data.isin}, AuNo {bond_data.auction_number}")
+                        action = "created"
                     
                     if bond_data.isin:
                         processed_isins.add(bond_data.isin)
+                    logger.info(
+                        "[bond-import:%s] auction=%s isin=%s %s, created=%s updated=%s failed=%s",
+                        task_id,
+                        bond_data.auction_number,
+                        bond_data.isin,
+                        action,
+                        created_count,
+                        updated_count,
+                        failed_count,
+                    )
 
                 except Exception as db_e:
                     failed_count += 1
-                    print(f"Database error for bond au_no {bond_data.auction_number}: {db_e}")
+                    logger.error(
+                        "[bond-import:%s] database error for auction=%s: %s",
+                        task_id,
+                        bond_data.auction_number,
+                        db_e,
+                    )
         
         summary = {
             "created": created_count,
@@ -153,10 +175,10 @@ async def run_bond_import_task(task_id: int):
             "message": "Bond import process finished."
         }
         await ImportTask.filter(id=task_id).update(status="completed", details=summary)
-        print(f"Bond import task {task_id} completed. Summary: {summary}")
+        logger.info("[bond-import:%s] completed: %s", task_id, summary)
 
     except Exception as e:
-        print(f"Fatal error in bond import task {task_id}: {e}")
+        logger.exception("[bond-import:%s] failed: %s", task_id, e)
         await ImportTask.filter(id=task_id).update(status="failed", details={"error": str(e)})
 
 

App/routers/bonds/utils.py

@@ -1,4 +1,5 @@
 import asyncio
+import logging
 from curl_cffi.requests import AsyncSession,RequestsError
 from bs4 import BeautifulSoup
 from App.routers.bonds.schemas import BondCreate # Adjust import path
@@ -9,6 +10,8 @@ import re
 from datetime import datetime as dt
 from typing import Tuple, Optional, List, Dict, Any
 
+logger = logging.getLogger(__name__)
+
 def parse_bond_title_details(title_str: str) -> Tuple[Optional[float], Optional[str], str, Optional[int]]:
     coupon_rate_val = None
     maturity_years_val = None
@@ -72,7 +75,7 @@ def parse_date_flexible(date_str: str, default=None) -> Optional[dt]:
             return dt.strptime(date_str, fmt).date()
         except ValueError:
             continue
-    print(f"Warning: Could not parse date string: {date_str}")
+    logger.warning("Could not parse date string: %s", date_str)
     return default
 
 def get_summary_item_value(summary_list: List[Dict[str, Any]], item_desc_key: str, default=None) -> Optional[str]:
@@ -116,9 +119,9 @@ class BondDataScraper:
             response.raise_for_status()
             return  response.json() if is_json else response.text
         except RequestsError as e: # Updated exception type for curl_cffi
-            print(f"HTTP error fetching {url}: {e}") # curl_cffi errors might not have response.status_code directly
+            logger.error("HTTP error fetching %s: %s", url, e) # curl_cffi errors might not have response.status_code directly
         except Exception as e:
-            print(f"Unexpected error fetching {url}: {e}")
+            logger.error("Unexpected error fetching %s: %s", url, e)
         return None
 
     async def _parse_main_tbonds_page(self, html_content: str) -> List[Dict[str, Any]]:
@@ -128,13 +131,13 @@ class BondDataScraper:
         soup = BeautifulSoup(html_content, 'html.parser')
         table = soup.find('table', class_='tbond-table') # Or id="DataTables_Table_0"
         if not table:
-            print("Main T-Bonds table not found.")
+            logger.warning("Main T-Bonds table not found.")
             return []
 
         parsed_rows = []
         tbody = table.find('tbody')
         if not tbody:
-            print("Tbody not found in main T-Bonds table.")
+            logger.warning("Tbody not found in main T-Bonds table.")
             return []
             
         for row in tbody.find_all('tr'):
@@ -150,7 +153,7 @@ class BondDataScraper:
                 
                 view_button = cols[4].find('button', id='showSummaryDetails')
                 if not view_button or not view_button.get('value'):
-                    print(f"Skipping row, view button or value not found for auction: {auction_number_text}")
+                    logger.warning("Skipping row, view button or value not found for auction: %s", auction_number_text)
                     continue
                 
                 button_value_parts = view_button['value'].split('_')
@@ -165,7 +168,7 @@ class BondDataScraper:
                     'au_days_part': au_days_part,
                 })
             except Exception as e:
-                print(f"Error parsing main table row: {e}. Row: {[c.get_text(strip=True) for c in cols]}")
+                logger.warning("Error parsing main table row: %s. Row: %s", e, [c.get_text(strip=True) for c in cols])
         return parsed_rows
 
     async def _fetch_bond_details(self, session: AsyncSession, au_no: int, au_days_part: str) -> Optional[Dict[str, Any]]:
@@ -174,7 +177,7 @@ class BondDataScraper:
 
     def _parse_bond_details_json(self, json_data: Dict[str, Any], initial_data: Dict[str, Any]) -> Optional[BondCreate]:
         if not json_data or json_data.get("message") != "SUCCESS":
-            print(f"Bond details JSON invalid or not successful for au_no: {initial_data.get('au_no')}")
+            logger.warning("Bond details JSON invalid or not successful for au_no: %s", initial_data.get("au_no"))
             return None
 
         summary_list = json_data.get("tbondSummary", [])
@@ -183,7 +186,7 @@ class BondDataScraper:
         
         auction_date_obj = parse_date_flexible(initial_data['table_auction_date_str'])
         if not auction_date_obj: # Critical, skip if no valid auction date
-             print(f"Critical: Could not parse auction_date for au_no: {initial_data.get('au_no')}")
+             logger.warning("Critical: could not parse auction_date for au_no: %s", initial_data.get("au_no"))
              return None
 
 
@@ -200,7 +203,7 @@ class BondDataScraper:
             try:
                 face_value_val = int(float(face_value_str.replace(",", "")) * 1_000_000)
             except ValueError:
-                print(f"Could not parse face_value: {face_value_str} for au_no: {initial_data.get('au_no')}")
+                logger.warning("Could not parse face_value %s for au_no: %s", face_value_str, initial_data.get("au_no"))
 
         price_per_100_str = get_summary_item_value(summary_list, "MINIMUM SUCCESSFUL PRICE / 100") # Or WAP?
         price_per_100_val = None
@@ -208,7 +211,7 @@ class BondDataScraper:
             try:
                 price_per_100_val = float(price_per_100_str)
             except ValueError:
-                 print(f"Could not parse price_per_100: {price_per_100_str} for au_no: {initial_data.get('au_no')}")
+                 logger.warning("Could not parse price_per_100 %s for au_no: %s", price_per_100_str, initial_data.get("au_no"))
 
 
         holding_number_str = json_data.get("auctionNumber") # This is the "1" from "AUCTION NUMBER 1 HELD ON..."
@@ -241,25 +244,42 @@ class BondDataScraper:
 
             main_page_html = await self._fetch_content(session, self.TBONDS_URL, method="GET")
             if not main_page_html:
-                print("Failed to fetch main T-Bonds page.")
+                logger.error("Failed to fetch main T-Bonds page.")
                 return
 
             initial_bond_rows = await self._parse_main_tbonds_page(main_page_html)
             
-            print(f"Found {len(initial_bond_rows)} initial bond rows from main table.")
+            total_rows = len(initial_bond_rows)
+            logger.info("[bond-scraper] main table loaded, rows=%s", total_rows)
 
-            for row_data in initial_bond_rows:
+            for index, row_data in enumerate(initial_bond_rows, start=1):
                 if row_data["au_no"] in existing_auction_numbers:
-                    print(f"Skipping existing bond details for au_no: {row_data['au_no']}")
+                    logger.info(
+                        "[bond-scraper] %s/%s auction=%s already stored, skipping detail fetch",
+                        index,
+                        total_rows,
+                        row_data["au_no"],
+                    )
                     continue
 
-                print(f"Fetching details for au_no: {row_data['au_no']}...")
+                logger.info(
+                    "[bond-scraper] %s/%s auction=%s fetching details",
+                    index,
+                    total_rows,
+                    row_data["au_no"],
+                )
                 await asyncio.sleep(0.5) # Small delay to be polite
                 
                 details_json = await self._fetch_bond_details(session, row_data['au_no'], row_data['au_days_part'])
                 if details_json:
                     bond_create_obj = self._parse_bond_details_json(details_json, row_data)
                     if bond_create_obj:
+                        logger.info(
+                            "[bond-scraper] %s/%s auction=%s parsed details",
+                            index,
+                            total_rows,
+                            row_data["au_no"],
+                        )
                         yield bond_create_obj
                 else:
-                    print(f"Failed to fetch or parse details for au_no: {row_data['au_no']}")
+                    logger.warning("Failed to fetch or parse details for au_no: %s", row_data["au_no"])

App/routers/funds/runner.py

@@ -7,6 +7,8 @@ Usage (from FastAPI BackgroundTasks):
 """
 import asyncio
 import logging
+import time
+import uuid
 from concurrent.futures import ThreadPoolExecutor
 from typing import List, Optional
 
@@ -69,13 +71,21 @@ def _get_scrapers(manager_name: str = "all") -> List[BaseFundScraper]:
 def _run_scraper_sync(scraper: BaseFundScraper) -> List[FundRecord]:
     """Run sync scraper; returns flat list of all FundRecord objects."""
     try:
+        logger.info("[fund-import] %s scrape started", scraper.manager_name)
+        started_at = time.monotonic()
         results = scraper.scrape_all()
         records: List[FundRecord] = []
         for fund_records in results.values():
             records.extend(fund_records)
+        logger.info(
+            "[fund-import] %s scrape finished, records=%s duration=%.1fs",
+            scraper.manager_name,
+            len(records),
+            time.monotonic() - started_at,
+        )
         return records
     except Exception as e:
-        logger.error(f"[{scraper.manager_name}] scrape failed: {e}")
+        logger.error("[fund-import] %s scrape failed: %s", scraper.manager_name, e)
         return []
 
 
@@ -98,7 +108,16 @@ async def _save_records(records: List[FundRecord], manager_name: str, website: s
         by_fund.setdefault(rec.fund_name, []).append(rec)
 
     # 3. For each fund, upsert MutualFund then bulk-insert new FundPerformance rows
-    for fund_name, fund_records in by_fund.items():
+    total_funds = len(by_fund)
+    for index, (fund_name, fund_records) in enumerate(by_fund.items(), start=1):
+        logger.info(
+            "[fund-import] %s save %s/%s %s started, records=%s",
+            manager_name,
+            index,
+            total_funds,
+            fund_name,
+            len(fund_records),
+        )
         meta = _FUND_META.get(fund_name, {})
         currency = meta.get("currency") or (fund_records[0].currency if fund_records else "TZS")
 
@@ -160,6 +179,15 @@ async def _save_records(records: List[FundRecord], manager_name: str, website: s
         if new_rows:
             await FundPerformance.bulk_create(new_rows, ignore_conflicts=True)
             stats["rows_added"] += len(new_rows)
+        logger.info(
+            "[fund-import] %s save %s/%s %s done, new=%s skipped_total=%s",
+            manager_name,
+            index,
+            total_funds,
+            fund_name,
+            len(new_rows),
+            stats["rows_skipped"],
+        )
 
     return stats
 
@@ -169,27 +197,71 @@ async def run_import(manager_name: str = "all") -> dict:
     Entry point called by FastAPI BackgroundTasks.
     Runs sync scrapers in a thread pool, then saves results to DB.
     """
+    run_id = uuid.uuid4().hex[:8]
+    started_at = time.monotonic()
     scrapers = _get_scrapers(manager_name)
     loop = asyncio.get_event_loop()
     all_stats: dict = {"managers": {}}
+    logger.info(
+        "[fund-import:%s] background import started, target=%s managers=%s",
+        run_id,
+        manager_name,
+        [scraper.manager_name for scraper in scrapers],
+    )
 
     with ThreadPoolExecutor(max_workers=len(scrapers)) as pool:
         futures = {
             loop.run_in_executor(pool, _run_scraper_sync, s): s
             for s in scrapers
         }
-        for future, scraper in futures.items():
+        total = len(futures)
+        for index, (future, scraper) in enumerate(futures.items(), start=1):
             try:
+                logger.info(
+                    "[fund-import:%s] manager %s/%s %s waiting for scrape result",
+                    run_id,
+                    index,
+                    total,
+                    scraper.manager_name,
+                )
                 records = await future
+                logger.info(
+                    "[fund-import:%s] manager %s/%s %s saving %s record(s)",
+                    run_id,
+                    index,
+                    total,
+                    scraper.manager_name,
+                    len(records),
+                )
                 stats = await _save_records(
                     records,
                     manager_name=scraper.manager_name,
                     website=getattr(scraper, "base_url", ""),
                 )
                 all_stats["managers"][scraper.manager_name] = stats
-                logger.info(f"[{scraper.manager_name}] import done: {stats}")
+                logger.info(
+                    "[fund-import:%s] manager %s/%s %s done: %s",
+                    run_id,
+                    index,
+                    total,
+                    scraper.manager_name,
+                    stats,
+                )
             except Exception as e:
-                logger.error(f"[{scraper.manager_name}] import error: {e}")
+                logger.error(
+                    "[fund-import:%s] manager %s/%s %s failed: %s",
+                    run_id,
+                    index,
+                    total,
+                    scraper.manager_name,
+                    e,
+                )
                 all_stats["managers"][scraper.manager_name] = {"error": str(e)}
 
+    logger.info(
+        "[fund-import:%s] background import finished in %.1fs: %s",
+        run_id,
+        time.monotonic() - started_at,
+        all_stats,
+    )
     return all_stats

App/scheduler.py

@@ -7,6 +7,8 @@ All external market checks run only three times per day:
 
 import asyncio
 import logging
+import time
+import uuid
 from datetime import datetime, timedelta
 from zoneinfo import ZoneInfo
 
@@ -16,6 +18,22 @@ MARKET_SYNC_TIMES = [(10, 0), (14, 0), (17, 0)]
 DAR_TZ = ZoneInfo("Africa/Dar_es_Salaam")
 
 
+async def _run_refresh_phase(run_id: str, name: str, refresh_func) -> None:
+    started_at = time.monotonic()
+    logger.info("[scheduler:%s] %s started", run_id, name)
+    try:
+        await refresh_func()
+    except Exception:
+        logger.exception("[scheduler:%s] %s failed", run_id, name)
+        return
+    logger.info(
+        "[scheduler:%s] %s finished in %.1fs",
+        run_id,
+        name,
+        time.monotonic() - started_at,
+    )
+
+
 async def refresh_stocks() -> None:
     """Refresh historical prices, current snapshot data, and corporate actions."""
     from App.routers.stocks.crud import (
@@ -30,10 +48,19 @@ async def refresh_stocks() -> None:
 
     stocks = await Stock.all()
     if stocks:
-        logger.info("[scheduler] Refreshing price history for %s stock(s)", len(stocks))
+        total = len(stocks)
+        logger.info("[scheduler] stocks: price history started for %s stock(s)", total)
         today = date_type.today()
-
-        for stock in stocks:
+        inserted_total = 0
+        failed_total = 0
+
+        for index, stock in enumerate(stocks, start=1):
+            logger.info(
+                "[scheduler] stocks: price history %s/%s %s started",
+                index,
+                total,
+                stock.symbol,
+            )
             try:
                 latest = (
                     await StockPriceData.filter(stock=stock).order_by("-date").first()
@@ -43,15 +70,15 @@ async def refresh_stocks() -> None:
                 data = await fetch_dse_stock_data(stock.symbol, days=days)
                 if not data.get("success"):
                     logger.warning(
-                        "[scheduler] %s: price-history API returned failure",
+                        "[scheduler] stocks: price history %s/%s %s API returned failure",
+                        index,
+                        total,
                         stock.symbol,
                     )
+                    failed_total += 1
                     continue
 
                 raw = data.get("data") or []
-                if not raw:
-                    continue
-
                 existing_dates = set(
                     await StockPriceData.filter(stock=stock).values_list(
                         "date", flat=True
@@ -66,37 +93,63 @@ async def refresh_stocks() -> None:
 
                 if new_rows:
                     await bulk_insert_price_data(stock, new_rows)
-                    logger.info(
-                        "[scheduler] %s: inserted %s new price record(s)",
-                        stock.symbol,
-                        len(new_rows),
-                    )
+                    inserted_total += len(new_rows)
+                logger.info(
+                    "[scheduler] stocks: price history %s/%s %s done, raw=%s new=%s",
+                    index,
+                    total,
+                    stock.symbol,
+                    len(raw),
+                    len(new_rows),
+                )
             except Exception as exc:
+                failed_total += 1
                 logger.error(
-                    "[scheduler] %s: price-history refresh failed: %s",
+                    "[scheduler] stocks: price history %s/%s %s failed: %s",
+                    index,
+                    total,
                     stock.symbol,
                     exc,
                 )
+        logger.info(
+            "[scheduler] stocks: price history complete, inserted=%s failed=%s",
+            inserted_total,
+            failed_total,
+        )
     else:
-        logger.info("[scheduler] No stocks in DB; skipping price-history refresh")
+        logger.info("[scheduler] stocks: no rows in DB; skipping price-history refresh")
 
     scraper = DseScraper()
 
     try:
+        logger.info("[scheduler] stocks: DSE market snapshot started")
         market_rows, _ = await scraper.fetch_market_snapshot()
         if market_rows:
             stats = await upsert_stock_reference_records(market_rows)
-            logger.info("[scheduler] DSE market snapshot sync complete: %s", stats)
+            logger.info(
+                "[scheduler] stocks: DSE market snapshot complete, rows=%s stats=%s",
+                len(market_rows),
+                stats,
+            )
+        else:
+            logger.info("[scheduler] stocks: DSE market snapshot complete, rows=0")
     except Exception as exc:
-        logger.error("[scheduler] DSE market snapshot sync failed: %s", exc)
+        logger.error("[scheduler] stocks: DSE market snapshot failed: %s", exc)
 
     try:
+        logger.info("[scheduler] stocks: DSE corporate actions started")
         actions, _ = await scraper.fetch_corporate_actions()
         if actions:
             stats = await upsert_corporate_actions(actions)
-            logger.info("[scheduler] DSE corporate actions sync complete: %s", stats)
+            logger.info(
+                "[scheduler] stocks: DSE corporate actions complete, rows=%s stats=%s",
+                len(actions),
+                stats,
+            )
+        else:
+            logger.info("[scheduler] stocks: DSE corporate actions complete, rows=0")
     except Exception as exc:
-        logger.error("[scheduler] DSE corporate actions sync failed: %s", exc)
+        logger.error("[scheduler] stocks: DSE corporate actions failed: %s", exc)
 
 
 async def refresh_funds() -> None:
@@ -104,10 +157,11 @@ async def refresh_funds() -> None:
     try:
         from App.routers.funds.runner import run_import
 
+        logger.info("[scheduler] funds: import started")
         stats = await run_import("all")
-        logger.info("[scheduler] Fund refresh complete: %s", stats)
+        logger.info("[scheduler] funds: import complete: %s", stats)
     except Exception as exc:
-        logger.error("[scheduler] Fund refresh failed: %s", exc)
+        logger.error("[scheduler] funds: import failed: %s", exc)
 
 
 async def refresh_bonds() -> None:
@@ -153,13 +207,24 @@ async def refresh_bonds() -> None:
                             **bond_data.dict(exclude_unset=True)
                         )
                         updated += 1
+                        action = "updated"
                     else:
                         await Bond.create(**bond_data.dict())
                         created += 1
                         existing_auction_numbers.add(bond_data.auction_number)
+                        action = "created"
 
                     if bond_data.isin:
                         processed_isins.add(bond_data.isin)
+                    logger.info(
+                        "[scheduler] bonds: auction=%s isin=%s %s, created=%s updated=%s failed=%s",
+                        bond_data.auction_number,
+                        bond_data.isin,
+                        action,
+                        created,
+                        updated,
+                        failed,
+                    )
                 except Exception as exc:
                     failed += 1
                     logger.error(
@@ -199,10 +264,21 @@ async def scheduled_market_loop() -> None:
     """Refresh stocks, funds, and bonds at the configured Dar es Salaam sync windows."""
     while True:
         delay = _seconds_until_next_market_sync()
-        logger.info("[scheduler] Next market sync in %.0f seconds", delay)
+        next_run = datetime.now(DAR_TZ) + timedelta(seconds=delay)
+        logger.info(
+            "[scheduler] Next market sync at %s, in %.0f seconds",
+            next_run.isoformat(timespec="seconds"),
+            delay,
+        )
         await asyncio.sleep(delay)
-        logger.info("[scheduler] Scheduled market sync triggered")
-        await refresh_stocks()
-        await refresh_funds()
-        await refresh_bonds()
-        logger.info("[scheduler] Scheduled market sync complete")
+        run_id = uuid.uuid4().hex[:8]
+        started_at = time.monotonic()
+        logger.info("[scheduler:%s] Scheduled market sync started", run_id)
+        await _run_refresh_phase(run_id, "stocks", refresh_stocks)
+        await _run_refresh_phase(run_id, "funds", refresh_funds)
+        await _run_refresh_phase(run_id, "bonds", refresh_bonds)
+        logger.info(
+            "[scheduler:%s] Scheduled market sync complete in %.1fs",
+            run_id,
+            time.monotonic() - started_at,
+        )

main.py

@@ -1,4 +1,5 @@
 import asyncio
+import logging
 from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
@@ -17,6 +18,12 @@ from App.scheduler import scheduled_market_loop
 
 from db import init_db, close_db, clear_db
 
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+)
+logger = logging.getLogger(__name__)
+
 app = FastAPI(title="Uwekezaji API", description="Stock Market Data API", redirect_slashes=False)
 
 
@@ -78,8 +85,11 @@ app.add_middleware(ProxyHeadersMiddleware, trusted_hosts="*")
 # Database initialization and cleanup
 @app.on_event("startup")
 async def startup_event():
+    logger.info("[startup] database initialization started")
     await init_db()
+    logger.info("[startup] database initialization complete")
     asyncio.create_task(scheduled_market_loop())
+    logger.info("[startup] scheduled market background loop started")
 
 
 @app.on_event("shutdown")