PKDevTools

---

Table of Contents

• What is PKDevTools?
• Installation
• Quick Start
• Architecture Overview
• Core Modules
  • Data Provider System
  • Logging Framework
  • Database Management
  • Environment & Configuration
  • Multiprocessing
  • Telegram Integration
  • GitHub Integration
  • Pub/Sub Event System
  • Utilities
• API Reference
• Environment Variables
• Contributing
• License

---

What is PKDevTools?

PKDevTools is a comprehensive Python toolkit designed for building high-performance financial applications. It provides:

• 🚀 Unified Data Provider - Multi-source stock data with automatic failover
• 📝 Thread-Safe Logging - Process-safe logging with filtering and caller info
• 🗄️ Database Management - SQLite + Turso (libsql) with sync capabilities
• ⚡ Multiprocessing - Cross-platform multiprocessing with shared state
• 📱 Telegram Integration - Send messages, documents, and media
• 🔄 GitHub Automation - Workflow triggers, commits, and API integration
• 📡 Event System - Pub/Sub pattern for decoupled components
• 🛠️ Utilities - Caching, archiving, HTTP fetching, and more

This toolkit serves as the foundation for PKScreener, PKBrokers, and PKNSETools.

---

Installation

From PyPI (Recommended)

pip install PKDevTools

From Source

git clone https://github.com/pkjmesra/PKDevTools.git
cd PKDevTools
pip install -r requirements.txt
pip install -e .

Requirements

• Python 3.9+
• See requirements.txt for full dependency list

---

Quick Start

fromPKDevTools.classesimportget_data_provider, get_scalable_fetcherfromPKDevTools.classes.logimportdefault_logger, setup_custom_logger# Initialize logging (set environment variable first)importosos.environ["PKDevTools_Default_Log_Level"] ="10"# DEBUG level# Get stock dataprovider=get_data_provider()
df=provider.get_stock_data("RELIANCE", interval="day", count=100)
# Use the loggerlogger=default_logger()
logger.info("Data fetched successfully!")

---

Architecture Overview

┌─────────────────────────────────────────────────────────────────────────┐
│ PKDevTools Architecture │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐ │
│ │ PKDataProvider │ │ PKScalableData │ │ DBManager │ │
│ │ (Stock Data) │ │ Fetcher (GitHub) │ │ (Turso/SQLite) │ │
│ └────────┬─────────┘ └────────┬─────────┘ └────────┬─────────┘ │
│ │ │ │ │
│ └─────────────────────┼─────────────────────┘ │
│ │ │
│ ┌────────────▼────────────┐ │
│ │ Core Services │ │
│ ├─────────────────────────┤ │
│ │ • Logging (filterlogger)│ │
│ │ • Environment Config │ │
│ │ • HTTP Fetcher │ │
│ │ • Archiver (Caching) │ │
│ └────────────┬────────────┘ │
│ │ │
│ ┌─────────────────────┼─────────────────────┐ │
│ │ │ │ │
│ ┌────────▼─────────┐ ┌────────▼───────┐ ┌──────────▼───────┐ │
│ │ Telegram │ │ GitHub │ │ Pub/Sub Events │ │
│ │ Integration │ │ Integration │ │ (blinker) │ │
│ └──────────────────┘ └────────────────┘ └──────────────────┘ │
│ │
│ ┌──────────────────────────────────────────────────────────────┐ │
│ │ Multiprocessing Layer │ │
│ │ PKMultiProcessorClient | PKJoinableQueue | Process Logging │ │
│ └──────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘

See Also: 1. Architecture 2. API Reference

---

Core Modules

1. Data Provider System

The unified data provider fetches stock OHLCV data from multiple sources with automatic failover.

PKDataProvider

fromPKDevTools.classes.PKDataProviderimportPKDataProvider, get_data_provider# Get singleton instanceprovider=get_data_provider()
# Fetch stock data with automatic source selection# Priority: Real-time (PKBrokers) → Local Pickle → Remote GitHub Pickledf=provider.get_stock_data("RELIANCE", interval="5m", count=50)
# Fetch multiple stocksdata=provider.get_multiple_stocks(["RELIANCE", "TCS", "INFY"], interval="day")
# Check real-time availabilityifprovider.is_realtime_available():
price=provider.get_latest_price("INFY")
ohlcv=provider.get_realtime_ohlcv("INFY")

Supported Intervals:

Interval	Description
1m, 2m, 3m, 4m, 5m	Minute candles
10m, 15m, 30m, 60m	Extended minute candles
day	Daily candles

PKScalableDataFetcher

GitHub-based data fetcher without Telegram dependency:

fromPKDevTools.classes.PKScalableDataFetcherimportPKScalableDataFetcher, get_scalable_fetcherfetcher=get_scalable_fetcher()
# Fetch from GitHub raw contentdata=fetcher.fetch_stock_data("RELIANCE")

---

2. Logging Framework

Thread and process-safe logging with automatic caller information injection.

Setup and Usage

importosfromPKDevTools.classes.logimport (
setup_custom_logger,
default_logger,
log_to,
tracelog
)
# Enable logging via environment variableos.environ["PKDevTools_Default_Log_Level"] ="10"# DEBUG=10, INFO=20, WARNING=30, ERROR=40# Setup custom loggerlogger=setup_custom_logger(
name="MyApp",
levelname=10, # DEBUGlog_file_path="/path/to/logs.txt",
filter="IMPORTANT"# Only log messages containing "IMPORTANT"
)
# Use default loggerlogger=default_logger()
logger.debug("Debug message")
logger.info("Info message")
logger.warning("Warning message")
logger.error("Error message") # Automatically includes tracebacklogger.critical("Critical message")

Decorator for Function Tracing

fromPKDevTools.classes.logimportlog_to, default_logger@log_to(default_logger().info)defmy_function(param1, param2):
"""Function calls are automatically logged with arguments and timing"""returnparam1+param2

Log Levels

Level	Value	Description
DEBUG	10	Detailed diagnostic information
INFO	20	General operational messages
WARNING	30	Warning messages
ERROR	40	Error messages with traceback
CRITICAL	50	Critical failures

Key Classes

• filterlogger: Thread/process-safe logger with filtering
• emptylogger: No-op logger when logging is disabled
• colors: ANSI color codes for terminal formatting

---

3. Database Management

Dual database support with SQLite (local) and Turso/libsql (cloud).

DBManager

fromPKDevTools.classes.DBManagerimportDBManager, PKUser# Initialize manager (uses environment variables for Turso connection)db=DBManager()
# User operationsuser=db.getUserByID(12345)
otp, subscription_model, validity, user=db.getOTP(
userID=12345,
userName="john_doe",
fullName="John Doe"
)
# Scanner job subscriptionsdb.subscribeScannerForUser(userID=12345, scannerIDs="X:12:9,X:12:31")
subscriptions=db.getSubscribedScannersByUser(userID=12345)

DatabaseSyncChecker

fromPKDevTools.classes.DatabaseSyncCheckerimportDatabaseSyncCheckerchecker=DatabaseSyncChecker(
local_db_path="./local.db",
turso_url="libsql://your-db.turso.io",
turso_auth_token="your-token"
)
needs_sync, messages=checker.check_sync_status()
checker.print_counts()

Key Models

• PKUser: User model with subscription management
• PKScannerJob: Scanner job subscription model
• PKUserModel: Enum for database column mapping

---

4. Environment & Configuration

Centralized environment variable and secrets management.

PKEnvironment

fromPKDevTools.classes.EnvironmentimportPKEnvironment# Singleton instance - loads from .env.dev fileenv=PKEnvironment()
# Access secrets as attributesgithub_token=env.GITHUB_TOKENchat_id=env.CHAT_IDtelegram_token=env.TOKEN# Access all secretsall_secrets=env.allSecrets# Returns dict

Required Environment Variables

Variable	Description
GITHUB_TOKEN	GitHub API token for repository operations
CHAT_ID	Telegram channel/chat ID
TOKEN	Telegram bot token
chat_idADMIN	Admin chat ID for notifications
PKDevTools_Default_Log_Level	Logging level (10=DEBUG, 20=INFO, etc.)

---

5. Multiprocessing

Cross-platform multiprocessing with shared state and logging support.

PKMultiProcessorClient

fromPKDevTools.classes.PKMultiProcessorClientimportPKMultiProcessorClientfromPKDevTools.classes.PKJoinableQueueimportPKJoinableQueuefrommultiprocessingimportManager# Create shared resourcesmanager=Manager()
task_queue=PKJoinableQueue()
result_queue=PKJoinableQueue()
# Define processor methoddefprocess_task(stock_code, data_dict, result_dict):
# Process stock dataresult=analyze_stock(stock_code)
returnresult# Create worker processesworkers= []
foriinrange(4): # 4 worker processesworker=PKMultiProcessorClient(
processorMethod=process_task,
task_queue=task_queue,
result_queue=result_queue,
objectDictionaryPrimary=manager.dict(),
keyboardInterruptEvent=manager.Event()
)
worker.start()
workers.append(worker)
# Add tasksforstockin ["RELIANCE", "TCS", "INFY"]:
task_queue.put(stock)
# Signal completion and waittask_queue.join()

PKJoinableQueue

Enhanced multiprocessing queue with join support:

fromPKDevTools.classes.PKJoinableQueueimportPKJoinableQueuequeue=PKJoinableQueue()
queue.put("task1")
queue.put("task2")
# Worker processes call task_done() after processingqueue.join() # Blocks until all tasks completed

---

6. Telegram Integration

Send messages, documents, and media to Telegram.

Basic Usage

fromPKDevTools.classes.Telegramimport (
send_message,
send_document,
send_photo,
send_media_group
)
# Send text messagesend_message(
message="Hello from PKDevTools!",
userID="-1001234567890",
parse_type="HTML"
)
# Send documentsend_document(
file_path="/path/to/file.pdf",
message="Here's your report",
userID="-1001234567890"
)
# Send photosend_photo(
photo_path="/path/to/image.png",
caption="Analysis results",
userID="-1001234567890"
)
# Send multiple documents as media groupsend_media_group(
file_paths=["/path/to/file1.pdf", "/path/to/file2.pdf"],
message="Multiple reports",
userID="-1001234567890"
)

Message Formatting

Messages support HTML formatting:

send_message(
message="<b>Bold</b> <i>Italic</i> <code>Code</code>",
userID=chat_id,
parse_type="HTML"
)

---

7. GitHub Integration

Automate GitHub operations including commits, workflow triggers, and API calls.

Committer

fromPKDevTools.classes.CommitterimportCommitter# Copy filesCommitter.copySourceToDestination(
srcPath="results/*.pkl",
destPath="backup/"
)
# Commit and push changesCommitter.commitTempOutcomes(
addPath="results/*",
commitMessage="[Auto] Updated results",
branchName="main"
)
# Execute OS command with loggingCommitter.execOSCommand("git status", showStatus=True)

WorkflowManager

fromPKDevTools.classes.WorkflowManagerimportWorkflowManager# Trigger GitHub Actions workflowWorkflowManager.trigger_workflow(
repo="pkjmesra/PKScreener",
workflow_id="scan.yml",
ref="main",
inputs={"scan_type": "full"}
)

githubutilities

fromPKDevTools.classes.githubutilitiesimport (
getWorkflowRunByName,
stopWorkflow,
getLatestRelease
)
# Get latest releaserelease=getLatestRelease("pkjmesra/PKScreener")
# Get workflow runrun=getWorkflowRunByName("pkjmesra/PKScreener", "Build")

---

8. Pub/Sub Event System

Decoupled event publishing and subscription using blinker.

Publishing Events

fromPKDevTools.classes.pubsub.publisherimportPKUserServicefromPKDevTools.classes.pubsub.eventsimportglobalEventsSignal# Using PKUserServiceservice=PKUserService()
service.notify_user(scannerID="X:12:9", notification="Scan complete!")
# Direct signal publishingglobalEventsSignal.send(
sender=self,
eventType="custom",
data={"key": "value"}
)

Subscribing to Events

fromPKDevTools.classes.pubsub.eventsimportglobalEventsSignaldefmy_handler(sender, **kwargs):
scanner_id=kwargs.get('scannerID')
notification=kwargs.get('notification')
print(f"Received: {scanner_id} - {notification}")
# Subscribe to eventsglobalEventsSignal.connect(my_handler)

---

9. Utilities

Archiver (Caching & File Management)

fromPKDevTools.classesimportArchiver# Get user data directorydata_dir=Archiver.get_user_data_dir()
# Get user outputs directoryoutputs_dir=Archiver.get_user_outputs_dir()
# Cache binary dataArchiver.cacheFile(binary_data, "cache_file.bin")
# Find cached filedata, path, modified_time=Archiver.findFile("cache_file.bin")
# Get last modified datetimemodified=Archiver.get_last_modified_datetime("/path/to/file")

Fetcher (HTTP Requests)

fromPKDevTools.classes.Fetcherimportfetcherf=fetcher()
# Fetch URL with cachingresponse=f.fetchURL("https://api.example.com/data")
# Fetch with custom headersresponse=f.fetchURL(
url="https://api.example.com/data",
headers={"Authorization": "Bearer token"}
)

PKDateUtilities

fromPKDevTools.classes.PKDateUtilitiesimportPKDateUtilities# Check if market is openis_open=PKDateUtilities.isTradingTime()
# Check if today is a holidayis_holiday=PKDateUtilities.isTradingHoliday()
# Get current IST timeist_now=PKDateUtilities.currentDateTime()
# Get trading day offsettrading_date=PKDateUtilities.tradingDate()

PKTimer

fromPKDevTools.classes.PKTimerimportPKTimer# Measure execution timewithPKTimer("Operation name"):
# Code to measureperform_operation()

ColorText

fromPKDevTools.classes.ColorTextimportcolorText# Print colored textprint(colorText.GREEN+"Success!"+colorText.END)
print(colorText.FAIL+"Error!"+colorText.END)
print(colorText.WARN+"Warning!"+colorText.END)

FunctionTimeouts

fromPKDevTools.classes.FunctionTimeoutsimportexit_after@exit_after(5) # Timeout after 5 secondsdefslow_function():
# Long running operationpass

---

API Reference

Main Exports

fromPKDevTools.classesimport (
# Data ProvidersPKDataProvider,
get_data_provider,
PKScalableDataFetcher,
get_scalable_fetcher,
# VersionVERSION,
)

Module Structure

PKDevTools/
├── classes/
│ ├── __init__.py # Main exports
│ ├── PKDataProvider.py # Unified data provider
│ ├── PKScalableDataFetcher.py # GitHub-based fetcher
│ ├── log.py # Logging framework
│ ├── DBManager.py # Database management
│ ├── Environment.py # Environment/secrets
│ ├── Fetcher.py # HTTP client
│ ├── Telegram.py # Telegram integration
│ ├── Committer.py # Git operations
│ ├── WorkflowManager.py # GitHub Actions
│ ├── PKMultiProcessorClient.py # Multiprocessing
│ ├── PKJoinableQueue.py # Enhanced queue
│ ├── Archiver.py # Caching/files
│ ├── PKDateUtilities.py # Date/time utilities
│ ├── pubsub/ # Event system
│ │ ├── events.py # Signal definitions
│ │ ├── publisher.py # Event publishing
│ │ └── subscriber.py # Event handling
│ └── ... # Other utilities
└── release.md # Release notes

---

Environment Variables

Variable	Required	Description
PKDevTools_Default_Log_Level	No	Logging level (10=DEBUG, 20=INFO, 30=WARNING, 40=ERROR)
GITHUB_TOKEN	Yes*	GitHub API token
TOKEN	Yes*	Telegram bot token
CHAT_ID	Yes*	Default Telegram chat ID
chat_idADMIN	No	Admin notification chat ID
TURSO_DB_URL	No	Turso database URL
TURSO_DB_AUTH_TOKEN	No	Turso authentication token

*Required for respective functionality

---

Contributing

We welcome contributions! Please follow these guidelines:

Development Setup

1. Fork the repository
2. Clone your fork:

   git clone https://github.com/YOUR_USERNAME/PKDevTools.git
   cd PKDevTools

3. Create a virtual environment:

   python -m venv venv
   source venv/bin/activate # or `venv\Scripts\activate` on Windows

4. Install development dependencies:

   pip install -r requirements.txt
   pip install -e .

Running Tests

# Run all tests
pytest test/
# Run with coverage
pytest --cov=PKDevTools test/
# Run specific test file
pytest test/DBManager_test.py

Code Style

We use ruff for linting:

ruff check PKDevTools/
ruff format PKDevTools/

Pull Request Guidelines

1. Create a feature branch from main
2. Write tests for new functionality
3. Ensure all tests pass
4. Update documentation as needed
5. Submit a pull request with a clear description

See CONTRIBUTING.md for detailed guidelines.

---

License

This project is licensed under the MIT License - see the LICENSE file for details.

---

Related Projects

• PKScreener - Stock screening application
• PKBrokers - Broker integration and real-time data
• PKNSETools - NSE market data tools

---

Support

• 📖 Documentation
• 🐛 Issue Tracker
• 💬 Discussions

---