Low Capacity Niche Screener¶

Comprehensive screening tool for identifying and analyzing low-capacity trading opportunities (TASK-RESEARCH-006).

Overview¶

The Low Capacity Niche Screener helps traders identify securities suitable for low-capacity strategies by analyzing liquidity constraints, market impact, and capacity limits. It's specifically designed for:

Small to medium trading operations
Niche market strategies
Strategies requiring low competition
Capacity-constrained alpha opportunities

Features¶

1. Volume Filtering¶

Filter securities by average daily volume (ADV)
Configurable min/max ADV thresholds in USD
Support for different liquidity tiers

2. Market Impact Estimation¶

Square-root model: Impact ∝ √(trade_size / ADV)
Linear model: Impact ∝ (trade_size / ADV)
Hybrid model: Combines both approaches
Incorporates spread and volatility components
Execution cost breakdown (impact, spread, total)

3. Capacity Calculation¶

Determine maximum tradeable position size
Account for multi-day holding periods
Support for multi-position portfolios
Optimal position sizing given impact constraints

4. Opportunity Ranking¶

Score opportunities by capacity/competition ratio
Weighted scoring across multiple dimensions:
Capacity score (higher = better)
Liquidity quality (lower spread/volatility = better)
Competition factor (lower ADV = less crowded)
Customizable scoring weights
Minimum score filtering

5. Alert Generation¶

New opportunities: Securities newly qualifying
Volume spikes: Significant ADV increases
Capacity changes: Material capacity shifts
Delistings: Securities no longer qualifying
Priority-based alert sorting

Installation¶

The module is part of the DSTA analytics package:

from src.analytics.capacity_screener import (
    MarketImpactModel,
    CapacityScreener,
    SecurityData,
    ImpactModelType,
    generate_screening_report
)

Quick Start¶

import numpy as np
from src.analytics.capacity_screener import (
    MarketImpactModel,
    CapacityScreener,
    SecurityData,
    ImpactModelType
)

# 1. Create securities
securities = [
    SecurityData(
        symbol='STOCK1',
        adv=50000,        # Average daily volume (shares)
        price=25.0,       # Current price
        volatility=0.30,  # Annualized volatility
        spread_bps=15.0   # Bid-ask spread in basis points
    ),
    # ... more securities
]

# 2. Configure impact model
impact_model = MarketImpactModel(
    model_type=ImpactModelType.SQUARE_ROOT,
    impact_coefficient=0.1,
    spread_coefficient=0.5,
    volatility_coefficient=0.03
)

# 3. Create screener
screener = CapacityScreener(
    impact_model=impact_model,
    min_adv_usd=100000,      # Minimum $100k ADV
    max_adv_usd=5000000,     # Maximum $5M ADV (low-capacity niche)
    max_impact_pct=1.0,      # 1% max impact
    target_holding_days=5    # 5-day holding period
)

# 4. Screen universe
results = screener.screen_universe(
    securities,
    rank_opportunities=True,
    generate_alerts=True,
    top_n=10
)

# 5. Access results
print(f"Qualifying securities: {len(results['qualifying_securities'])}")
for opp in results['ranked_opportunities']:
    print(f"{opp.symbol}: Score={opp.score:.1f}, Capacity=${opp.capacity:,.0f}")

Classes and Methods¶

`SecurityData`¶

Dataclass representing security information.

Attributes: - symbol: Security ticker - adv: Average daily volume (shares) - price: Current price - volatility: Annualized volatility (optional) - spread_bps: Bid-ask spread in basis points (optional) - market_cap: Market capitalization (optional)

Properties: - adv_notional: ADV in dollar terms (ADV × price)

`MarketImpactModel`¶

Estimates market impact for trades.

Methods: - calculate_impact(position_size_usd, security): Calculate impact percentage - calculate_optimal_position_size(security, max_impact_pct): Find optimal size - estimate_execution_cost(position_size_usd, security, side): Full cost breakdown

Impact Models: - SQUARE_ROOT: Sublinear impact (default) - LINEAR: Linear impact - HYBRID: Adaptive (sqrt for small, linear for large)

`CapacityScreener`¶

Main screening and analysis tool.

Key Methods:

`filter_by_adv(securities, min_adv_usd, max_adv_usd)`¶

Filter securities by ADV range.

qualified = screener.filter_by_adv(
    securities,
    min_adv_usd=100000,
    max_adv_usd=10000000
)

`calculate_strategy_capacity(security, max_impact_pct, num_positions)`¶

Calculate maximum tradeable size.

capacity = screener.calculate_strategy_capacity(
    security,
    max_impact_pct=1.0,
    num_positions=10
)
# Returns: {
#     'single_position_capacity_usd': 500000,
#     'position_capacity_with_time_usd': 2500000,
#     'total_strategy_capacity_usd': 25000000,
#     'daily_participation_rate': 0.05,
#     ...
# }

`rank_opportunities(securities, weight_capacity, weight_liquidity, weight_competition)`¶

Rank and score opportunities.

ranked = screener.rank_opportunities(
    securities,
    weight_capacity=0.3,
    weight_liquidity=0.3,
    weight_competition=0.4,
    min_score=50.0
)

`generate_alerts(securities, volume_spike_threshold, capacity_change_threshold)`¶

Generate alerts for changes.

alerts = screener.generate_alerts(
    current_securities,
    volume_spike_threshold=2.0,      # 2x volume = spike
    capacity_change_threshold=0.5    # 50% change = alert
)

`screen_universe(securities, rank_opportunities, generate_alerts, top_n)`¶

Complete screening workflow.

results = screener.screen_universe(
    securities,
    rank_opportunities=True,
    generate_alerts=True,
    top_n=20
)

Returns:

{
    'qualifying_securities': [...],
    'ranked_opportunities': [...],
    'alerts': [...],
    'summary': {
        'total_securities': 100,
        'qualifying_securities': 25,
        'qualification_rate': 25.0,
        'avg_adv_usd': 2500000,
        'total_capacity_usd': 50000000,
        ...
    },
    'parameters': {...}
}

`export_results(results, filepath, format)`¶

Export results to file.

screener.export_results(results, 'screening.csv', format='csv')
screener.export_results(results, 'screening.json', format='json')
screener.export_results(results, 'screening.xlsx', format='excel')

Advanced Usage¶

Custom Impact Model¶

# Create custom hybrid model
impact_model = MarketImpactModel(
    model_type=ImpactModelType.HYBRID,
    impact_coefficient=0.08,      # Lower for liquid markets
    spread_coefficient=0.5,       # Half-spread assumption
    volatility_coefficient=0.03   # Vol adjustment
)

Multi-Period Screening¶

# Period 1
results_t1 = screener.screen_universe(securities_t1)

# Period 2 - detect changes
results_t2 = screener.screen_universe(
    securities_t2,
    volume_spike_threshold=2.0,
    capacity_change_threshold=0.3
)

# Check alerts
for alert in results_t2['alerts']:
    if alert.alert_type == 'volume_spike':
        print(f"Volume spike in {alert.symbol}!")

Custom Scoring¶

# Emphasize low competition
ranked = screener.rank_opportunities(
    securities,
    weight_capacity=0.2,
    weight_liquidity=0.1,
    weight_competition=0.7,  # High weight on low competition
    min_score=60.0
)

Generate Reports¶

from src.analytics.capacity_screener import generate_screening_report

report = generate_screening_report(results, screener)
print(report)

# Or save to file
with open('screening_report.txt', 'w') as f:
    f.write(report)

Examples¶

See examples/capacity_screening_demo.py for a complete working example.

Run the demo:

cd /home/minhdqdev/Projects/dsta
python examples/capacity_screening_demo.py

Testing¶

Comprehensive test suite with 37 tests covering: - Data validation - Market impact calculations - Capacity estimation - Opportunity ranking - Alert generation - File export - Integration workflows

Run tests:

pytest tests/analytics/test_capacity_screener.py -v -p no:django

Test Results: - ✅ 35 tests passed - ⏭️ 2 tests skipped (require openpyxl) - ⚡ Runs in ~1.5 seconds

Implementation Notes¶

Market Impact Models¶

The screener implements three impact models:

Square-root model (default):
Impact = coef × √(participation_rate) + spread + vol_component
Best for most liquid markets
Sublinear scaling suitable for typical market impact
Linear model:
Impact = coef × participation_rate + spread + vol_component
More conservative for illiquid markets
Linear scaling for safety
Hybrid model:
Uses sqrt below 10% participation
Uses linear above 10% participation
Adaptive approach for varying trade sizes

Capacity Calculation¶

Capacity is calculated considering: - Single position max size (constrained by impact) - Multi-day accumulation (holding_days × single_position) - Multi-position portfolio (num_positions × position_capacity)

Formula:

total_capacity = optimal_position_size × holding_days × num_positions

Where optimal_position_size is found via binary search to satisfy:

impact(optimal_position_size) = max_impact_pct

Opportunity Scoring¶

Opportunities are scored (0-100) using weighted components:

Capacity Score (0-100):
Log-scale normalized
Higher capacity = higher score
Liquidity Score (0-100):
Penalized by spread (up to -50 points)
Penalized by volatility (up to -30 points)
Higher quality = higher score
Competition Score (0-100):
Based on ADV within range
Lower ADV = less competition = higher score

Final score:

score = w_cap × cap_score + w_liq × liq_score + w_comp × comp_score

Performance Considerations¶

Screening 1000 securities: ~0.5 seconds
Memory efficient: processes securities in single pass
Alert generation: O(n) complexity
Ranking: O(n log n) for sorting

Error Handling¶

The module includes comprehensive error handling: - Validates ADV and price are positive - Handles edge cases (empty universes, no qualifying securities) - Graceful degradation for missing optional fields - Warnings for empty exports

Future Enhancements¶

Potential improvements: - [ ] Real-time data integration - [ ] Historical capacity analysis - [ ] Correlation-adjusted capacity - [ ] Sector/industry grouping - [ ] Machine learning-based impact models - [ ] Multi-currency support - [ ] Backtesting integration

Author¶

DSTA Team
Date: 2026-01-27
Task: TASK-RESEARCH-006

License¶

Internal use only - DSTA Project