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

from PKDevTools.classes import get_data_provider, get_scalable_fetcher
from PKDevTools.classes.log import default_logger, setup_custom_logger

# Initialize logging (set environment variable first)
import os
os.environ["PKDevTools_Default_Log_Level"] = "10"  # DEBUG level

# Get stock data
provider = get_data_provider()
df = provider.get_stock_data("RELIANCE", interval="day", count=100)

# Use the logger
logger = 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   │       │
│  └──────────────────────────────────────────────────────────────┘       │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘

Core Modules

1. Data Provider System

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

PKDataProvider

from PKDevTools.classes.PKDataProvider import PKDataProvider, get_data_provider

# Get singleton instance
provider = get_data_provider()

# Fetch stock data with automatic source selection
# Priority: Real-time (PKBrokers) → Local Pickle → Remote GitHub Pickle
df = provider.get_stock_data("RELIANCE", interval="5m", count=50)

# Fetch multiple stocks
data = provider.get_multiple_stocks(["RELIANCE", "TCS", "INFY"], interval="day")

# Check real-time availability
if provider.is_realtime_available():
    price = provider.get_latest_price("INFY")
    ohlcv = provider.get_realtime_ohlcv("INFY")

Supported Intervals:

PKScalableDataFetcher

GitHub-based data fetcher without Telegram dependency:

from PKDevTools.classes.PKScalableDataFetcher import PKScalableDataFetcher, get_scalable_fetcher

fetcher = get_scalable_fetcher()

# Fetch from GitHub raw content
data = fetcher.fetch_stock_data("RELIANCE")

2. Logging Framework

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

Setup and Usage

import os
from PKDevTools.classes.log import (
    setup_custom_logger,
    default_logger,
    log_to,
    tracelog
)

# Enable logging via environment variable
os.environ["PKDevTools_Default_Log_Level"] = "10"  # DEBUG=10, INFO=20, WARNING=30, ERROR=40

# Setup custom logger
logger = setup_custom_logger(
    name="MyApp",
    levelname=10,  # DEBUG
    log_file_path="/path/to/logs.txt",
    filter="IMPORTANT"  # Only log messages containing "IMPORTANT"
)

# Use default logger
logger = default_logger()
logger.debug("Debug message")
logger.info("Info message")
logger.warning("Warning message")
logger.error("Error message")  # Automatically includes traceback
logger.critical("Critical message")

Decorator for Function Tracing

from PKDevTools.classes.log import log_to, default_logger

@log_to(default_logger().info)
def my_function(param1, param2):
    """Function calls are automatically logged with arguments and timing"""
    return param1 + param2

Log Levels

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

from PKDevTools.classes.DBManager import DBManager, PKUser

# Initialize manager (uses environment variables for Turso connection)
db = DBManager()

# User operations
user = db.getUserByID(12345)
otp, subscription_model, validity, user = db.getOTP(
    userID=12345,
    userName="john_doe",
    fullName="John Doe"
)

# Scanner job subscriptions
db.subscribeScannerForUser(userID=12345, scannerIDs="X:12:9,X:12:31")
subscriptions = db.getSubscribedScannersByUser(userID=12345)

DatabaseSyncChecker

from PKDevTools.classes.DatabaseSyncChecker import DatabaseSyncChecker

checker = 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

from PKDevTools.classes.Environment import PKEnvironment

# Singleton instance - loads from .env.dev file
env = PKEnvironment()

# Access secrets as attributes
github_token = env.GITHUB_TOKEN
chat_id = env.CHAT_ID
telegram_token = env.TOKEN

# Access all secrets
all_secrets = env.allSecrets  # Returns dict

Required Environment Variables

5. Multiprocessing

Cross-platform multiprocessing with shared state and logging support.

PKMultiProcessorClient

from PKDevTools.classes.PKMultiProcessorClient import PKMultiProcessorClient
from PKDevTools.classes.PKJoinableQueue import PKJoinableQueue
from multiprocessing import Manager

# Create shared resources
manager = Manager()
task_queue = PKJoinableQueue()
result_queue = PKJoinableQueue()

# Define processor method
def process_task(stock_code, data_dict, result_dict):
    # Process stock data
    result = analyze_stock(stock_code)
    return result

# Create worker processes
workers = []
for i in range(4):  # 4 worker processes
    worker = PKMultiProcessorClient(
        processorMethod=process_task,
        task_queue=task_queue,
        result_queue=result_queue,
        objectDictionaryPrimary=manager.dict(),
        keyboardInterruptEvent=manager.Event()
    )
    worker.start()
    workers.append(worker)

# Add tasks
for stock in ["RELIANCE", "TCS", "INFY"]:
    task_queue.put(stock)

# Signal completion and wait
task_queue.join()

PKJoinableQueue

Enhanced multiprocessing queue with join support:

from PKDevTools.classes.PKJoinableQueue import PKJoinableQueue

queue = PKJoinableQueue()
queue.put("task1")
queue.put("task2")

# Worker processes call task_done() after processing
queue.join()  # Blocks until all tasks completed

6. Telegram Integration

Send messages, documents, and media to Telegram.

Basic Usage

from PKDevTools.classes.Telegram import (
    send_message,
    send_document,
    send_photo,
    send_media_group
)

# Send text message
send_message(
    message="Hello from PKDevTools!",
    userID="-1001234567890",
    parse_type="HTML"
)

# Send document
send_document(
    file_path="/path/to/file.pdf",
    message="Here's your report",
    userID="-1001234567890"
)

# Send photo
send_photo(
    photo_path="/path/to/image.png",
    caption="Analysis results",
    userID="-1001234567890"
)

# Send multiple documents as media group
send_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

from PKDevTools.classes.Committer import Committer

# Copy files
Committer.copySourceToDestination(
    srcPath="results/*.pkl",
    destPath="backup/"
)

# Commit and push changes
Committer.commitTempOutcomes(
    addPath="results/*",
    commitMessage="[Auto] Updated results",
    branchName="main"
)

# Execute OS command with logging
Committer.execOSCommand("git status", showStatus=True)

WorkflowManager

from PKDevTools.classes.WorkflowManager import WorkflowManager

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

githubutilities

from PKDevTools.classes.githubutilities import (
    getWorkflowRunByName,
    stopWorkflow,
    getLatestRelease
)

# Get latest release
release = getLatestRelease("pkjmesra/PKScreener")

# Get workflow run
run = getWorkflowRunByName("pkjmesra/PKScreener", "Build")

8. Pub/Sub Event System

Decoupled event publishing and subscription using blinker.

Publishing Events

from PKDevTools.classes.pubsub.publisher import PKUserService
from PKDevTools.classes.pubsub.events import globalEventsSignal

# Using PKUserService
service = PKUserService()
service.notify_user(scannerID="X:12:9", notification="Scan complete!")

# Direct signal publishing
globalEventsSignal.send(
    sender=self,
    eventType="custom",
    data={"key": "value"}
)

Subscribing to Events

from PKDevTools.classes.pubsub.events import globalEventsSignal

def my_handler(sender, **kwargs):
    scanner_id = kwargs.get('scannerID')
    notification = kwargs.get('notification')
    print(f"Received: {scanner_id} - {notification}")

# Subscribe to events
globalEventsSignal.connect(my_handler)

9. Utilities

Archiver (Caching & File Management)

from PKDevTools.classes import Archiver

# Get user data directory
data_dir = Archiver.get_user_data_dir()

# Get user outputs directory
outputs_dir = Archiver.get_user_outputs_dir()

# Cache binary data
Archiver.cacheFile(binary_data, "cache_file.bin")

# Find cached file
data, path, modified_time = Archiver.findFile("cache_file.bin")

# Get last modified datetime
modified = Archiver.get_last_modified_datetime("/path/to/file")

Fetcher (HTTP Requests)

from PKDevTools.classes.Fetcher import fetcher

f = fetcher()

# Fetch URL with caching
response = f.fetchURL("https://api.example.com/data")

# Fetch with custom headers
response = f.fetchURL(
    url="https://api.example.com/data",
    headers={"Authorization": "Bearer token"}
)

PKDateUtilities

from PKDevTools.classes.PKDateUtilities import PKDateUtilities

# Check if market is open
is_open = PKDateUtilities.isTradingTime()

# Check if today is a holiday
is_holiday = PKDateUtilities.isTradingHoliday()

# Get current IST time
ist_now = PKDateUtilities.currentDateTime()

# Get trading day offset
trading_date = PKDateUtilities.tradingDate()

PKTimer

from PKDevTools.classes.PKTimer import PKTimer

# Measure execution time
with PKTimer("Operation name"):
    # Code to measure
    perform_operation()

ColorText

from PKDevTools.classes.ColorText import colorText

# Print colored text
print(colorText.GREEN + "Success!" + colorText.END)
print(colorText.FAIL + "Error!" + colorText.END)
print(colorText.WARN + "Warning!" + colorText.END)

FunctionTimeouts

from PKDevTools.classes.FunctionTimeouts import exit_after

@exit_after(5)  # Timeout after 5 seconds
def slow_function():
    # Long running operation
    pass

API Reference

Main Exports

from PKDevTools.classes import (
    # Data Providers
    PKDataProvider,
    get_data_provider,
    PKScalableDataFetcher,
    get_scalable_fetcher,
    
    # Version
    VERSION,
)

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

*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