How dhivkrish7/lkfinance-engine works
1. System purpose and shape
- lkfinance is a Python library plus two operational entry points: a data-ingestion collector engine and a FastAPI read API.
- The core domain is Colombo Stock Exchange market data: tickers, sectors, indices, market summaries, and trade summaries.
2. External data source boundary
- The only clearly implied upstream dependency is the CSE web/data source.
- All live market data enters the system through HTTP, then is parsed and normalized before becoming library objects or stored records.
3. HTTP client layer
- lkfinance.client.CSEClient is the low-level outbound HTTP boundary.
- It owns request sessions, retries, timeouts, logging, JSON parsing, and HTTP error handling, so higher layers do not talk to requests directly.
- This layer is reused by collectors and API-facing services.
4. API contract and parsing boundary
- lkfinance.api defines endpoint names, response parsing, middleware, and route wiring.
- lkfinance.api.services.market_data sits between route handlers and the lower-level client/data logic.
- lkfinance.api.parser, schemas.py, and routes/* separate wire-format concerns from domain objects.
5. Domain model layer
- lkfinance.market contains the user-facing objects: Ticker, Index, Sector, Summary, and Lookup.
- lkfinance.models contains immutable dataclass-style data structures such as ticker/sector/summary/index data.
- This is the main boundary between raw API payloads and stable library consumers.
6. Normalization, validation, and parsing
- lkfinance.parsers and lkfinance.validators isolate the messy parts of turning CSE responses into consistent values.
- lkfinance.parsers.cse and normalizers.py suggest source-specific parsing and cleanup.
- lkfinance.exceptions provides explicit failure modes for invalid or incomplete market data.
7. Collector subsystem
- lkfinance.collectors is the ingestion side of the system.
- It includes collectors for tickers, sectors, indices, market summary, and trade summary, plus a summary_pipeline and engine.
- These collectors are reusable jobs that fetch data, transform it, and hand it to storage; they are not tied to the FastAPI layer.
8. Persistence and storage boundary
- lkfinance.database owns SQLAlchemy ORM models, session management, repositories, and adapters for SQLite, PostgreSQL, and Redis cache.
- lkfinance.storage appears to be a lighter storage abstraction for collector outputs, with SQLite as the default implementation.
- The architecture keeps collectors independent from the concrete database engine by routing writes through repository/storage interfaces.
9. Caching and optional speed layer
- lkfinance.cache and lkfinance.database.redis_cache indicate an optional cache layer, likely used to reduce repeated fetches or reads.
- Cache is a supporting subsystem, not the primary source of truth; persistence is still centered on the database/storage adapters.
10. Scheduler and batch execution
- lkfinance.scheduler.CollectorScheduler uses APScheduler to run collectors on intervals or cron schedules.
- It includes overlap prevention and graceful shutdown, so collector jobs can run as long-lived background tasks without clashing.
- This is the orchestration layer for continuous ingestion.
11. End-to-end data flow
- Typical ingestion flow: CSE HTTP response -> CSEClient -> parser/validator/normalizer -> market model/dataclass -> collector -> storage/database.
- Typical read flow for the API: FastAPI route -> service/repository -> database/storage -> serialized response.
- The library API likely sits on top of the same domain models, exposing convenient constructors like lk.Ticker(...) and lk.Summary(...).
12. Runtime surfaces
- run_engine.py is a minimal standalone collector prototype that fetches market summary data and writes to SQLite.
- uvicorn lkfinance.api.app:app runs the REST API.
- python -m lkfinance.scheduler runs the scheduled collector process.
13. Testing and package boundaries
- Tests cover the API, collectors, scheduler, sector, summary, and ticker flows, which mirrors the package boundaries well.
- The repository is packaged as a typed Python distribution (py.typed, pyproject.toml, setup.py), so type boundaries are part of the design rather than an afterthought.
Main entry points: market_data.py Market service · summary_pipeline.py Summary pipeline · repositories.py Repositories · cse.py CSE parser
Components 23
API
FastAPI read surface, contracts, and service layer.
- API app · Application assembly and startup for the read API.
lkfinance/api/app.py - Market route · Market data endpoint wiring for the API.
lkfinance/api/routes/market.py - Market service · Bridges HTTP handlers to market-data retrieval and transformation.
lkfinance/api/services/market_data.py - API parser · Wire-format parsing and endpoint response handling.
lkfinance/api/parser.py - Schemas · API request/response schemas and serialization shapes.
lkfinance/api/schemas.py
Ingestion
Collectors, pipelines, and scheduler for market data ingestion.
- Collector engine · Coordinates collector execution and job flow.
lkfinance/collectors/engine.py - Summary pipeline · Transforms market summary data through collector stages.
lkfinance/collectors/summary_pipeline.py - Collector base · Shared collector behavior used by ingestion jobs.
lkfinance/collectors/base.py
Domain
Stable market objects, models, parsers, and validation.
- CSE parser · Source-specific parsing of CSE responses.
lkfinance/parsers/cse.py - Normalizers · Canonicalization and cleanup of raw market fields.
lkfinance/parsers/normalizers.py - CSE validators · Checks completeness and validity of market payloads.
lkfinance/validators/cse.py - Market objects · User-facing ticker, sector, index, summary, and lookup types.
lkfinance/market/__init__.py - Core models · Dataclass-style data structures for tickers, sectors, summaries, and indices.
lkfinance/models/__init__.py
Persistence
Storage, ORM models, repositories, sessions, and cache adapters.
- Storage base · Abstract storage boundary for collector outputs.
lkfinance/storage/base.py - SQLite storage · Default lightweight storage implementation.
lkfinance/storage/sqlite.py - Repositories · Persistence-facing repositories for read and write access.
lkfinance/database/repositories.py - DB session · Database session creation and lifecycle management.
lkfinance/database/session.py - ORM models · SQLAlchemy-backed persistence models.
lkfinance/database/models.py - Redis cache · Optional Redis-backed cache for repeated reads or fetches.
lkfinance/database/redis_cache.py
Shared
Reusable cross-cutting client, cache, and configuration pieces.
- CSE client · Outbound HTTP boundary to the Colombo Stock Exchange data source.
lkfinance/client.py - Cache · Shared cache layer used by higher-level components.
lkfinance/cache.py
Runtime
Standalone entry points for API and batch execution.
- Scheduler · Runs collectors on intervals or cron schedules.
lkfinance/scheduler.py - Engine runner · Standalone collector prototype for market summary ingestion.
run_engine.py
Connections 24
- API app → Market route: mounts · Registers market endpoints on the application.
- Market route → Market service: calls · Route handlers delegate to the market service.
- Market service → API parser: parses · Service layer uses parser helpers for response shaping.
- Market service → CSE client: fetches · Service can reach live market data through the shared client boundary.
- Market service → Repositories: reads · Service can source persisted market data through repositories.
- API parser → Schemas: maps to · Parsed payloads are converted into API schemas.
- CSE client → CSE parser: feeds · Raw HTTP payloads are handed to CSE-specific parsing.
- CSE parser → Normalizers: cleans · Source fields are normalized into consistent values.
- CSE parser → CSE validators: checks · Parsed payloads are validated before becoming domain data.
- CSE validators → Core models: produces · Validated data becomes stable model objects.
- Core models → Market objects: wraps · Core models back the higher-level market objects.
- Collector engine → Collector base: orchestrates · Engine coordinates collector execution via shared base behavior.
- Collector engine → Summary pipeline: runs · Engine can invoke the summary pipeline as part of ingestion.
- Summary pipeline → CSE client: fetches · Pipeline uses the shared HTTP client to obtain live data.
- Summary pipeline → Core models: emits · Pipeline converts fetched data into durable model records.
- Summary pipeline → Storage base: writes · Pipeline hands results to the storage abstraction.
- SQLite storage → Storage base: implements · SQLite is the default storage implementation.
- Repositories → DB session: uses · Repositories operate through managed DB sessions.
- Repositories → ORM models: persists · Repositories translate domain state into ORM models.
- Repositories → Redis cache: accelerates · Repositories may consult Redis for cached values.
- Cache → Redis cache: adapts · Shared cache facade can route to Redis-backed storage.
- Scheduler → Collector engine: schedules · Scheduler triggers collector execution on a cadence.
- Engine runner → Summary pipeline: invokes · Standalone runner executes the market-summary pipeline.
- Market service → Cache: consults · API reads may use the shared cache layer for faster access.