lkfinance

lkfinance is a Python library for Colombo Stock Exchange market data. It is structured like a production package: a reusable API client, typed market objects, dataclass models, explicit exceptions, and pluggable cache/storage adapters.

Installation

pip install lkfinance

Usage

import lkfinance as lk

aspi = lk.aspi()
print(aspi.price)
print(aspi.percentage)
print(aspi.to_dict())

ticker = lk.Ticker("JKH")
print(ticker.name, ticker.price)

summary = lk.Summary("market")
print(summary.market_cap)
print(summary.metrics)

Architecture

• lkfinance.client.CSEClient owns requests.Session, retries, timeouts, logging, JSON parsing, and HTTP exceptions.
• lkfinance.api owns endpoint names and response parsers.
• lkfinance.market owns user-facing objects such as Ticker, Index, Sector, Summary, and Lookup.
• lkfinance.models owns immutable dataclasses such as TickerData, SectorData, IndexData, and SummaryData.
• lkfinance.database owns SQLAlchemy ORM models, session management, repositories, and optional SQLite, Redis, and PostgreSQL adapters.
• lkfinance.collectors owns reusable ingestion collectors for market summary, tickers, sectors, indices, and trade summary snapshots.
• lkfinance.parsers, lkfinance.validators, and lkfinance.storage separate parsing, validation, normalization, and persistence concerns.
• lkfinance.scheduler.CollectorScheduler uses APScheduler to run collectors independently with interval or cron triggers, overlap prevention, structured logs, and graceful shutdown.

Collector Engine

Minimal end-to-end market summary prototype:

python run_engine.py

This fetches the CSE market summary with requests.Session, creates lkfinance_engine.sqlite3 if missing, stores timestamp, market cap, turnover, ASPI, S&P SL20, and trade count, logs the process, and verifies insertion.

from lkfinance import CollectorEngine, SQLiteMarketDataStore

store = SQLiteMarketDataStore("market_data.sqlite3")
engine = CollectorEngine(storage=store)
results = engine.run()

for result in results:
    print(result.collector, result.success, result.records_saved)

from lkfinance import CollectorScheduler, TickerCollector

scheduler = CollectorScheduler()
scheduler.add(TickerCollector(["JKH"], storage=store), interval_seconds=60)
scheduler.start()

Production-style collector scheduler:

python -m lkfinance.scheduler

This starts the ticker collector every 10 seconds, plus sector and market summary collectors every 5 minutes. Jobs use max_instances=1 to prevent overlap and can also be scheduled with cron triggers:

scheduler.add_cron(TickerCollector(["JKH"], storage=store), minute="*/5")

The default storage adapter uses SQLite through SQLAlchemy and creates tables automatically. The database layer is split into ORM models (lkfinance.database.models), engine/session management (lkfinance.database.session), and repositories (lkfinance.database.repositories) so a PostgreSQL URL can be introduced without changing collector code.

FastAPI Backend

Run the REST API against the default SQLite market-data database:

uvicorn lkfinance.api.app:app --reload

Available endpoints include:

• GET /health
• GET /api/v1/health
• GET /api/v1/ticker/JKH
• GET /api/v1/market/summary
• GET /api/v1/sectors
• GET /api/v1/indices

Set LKFINANCE_DATABASE_URL to point the API at another SQLAlchemy database URL, or LKFINANCE_DATABASE_PATH for a different SQLite file.

Testing

python -m unittest discover -s tests -v

About

Author - Dhivyarajan K'Yozhandren
GitHub - DhivKrish7