max-concurrent-positions

name	max-concurrent-positions
description	Use before opening any new position to determine whether adding it would violate concurrent position limits. Enforces per-strategy, per-direction, per-correlation-cluster, and total portfolio limits. Integrates kill-switch tier state and regime conditions to dynamically tighten caps. Always consult this skill after kelly-position-sizing-perps and before slippage-budget-enforcement.
category	agentic

Max Concurrent Positions

When This Skill Activates

Apply this skill:

• Immediately after kelly-position-sizing-perps produces a sized notional — before any order is submitted
• When evaluating whether a second leg or scale-in is permitted on an existing position
• At session open to audit the inherited position count and confirm caps are not already breached from a prior session
• After any kill-switch tier change — tier changes dynamically tighten all caps and may require immediate position reduction
• When a new strategy is added to the active strategy set — check whether combined capacity of all strategies fits within portfolio caps
• During regime transitions — trending → ranging or low → high cascade score changes the effective caps

---

Cap Hierarchy

Concurrent position limits are enforced at five nested levels. A new position must clear all five before entry is permitted. The binding cap is the one that produces the smallest allowed count at the moment of evaluation.

Level 1: Total portfolio positions          (global ceiling)
Level 2: Per-direction (long / short)       (directional concentration)
Level 3: Per-correlation-cluster            (correlated exposure grouping)
Level 4: Per-strategy                       (single strategy capacity)
Level 5: Per-asset                          (repeat position guard)

---

Default Cap Table

Caps are defined in kill_switch_config.yaml and dynamically overridden by kill-switch tier and regime state.

Cap Level	Normal	Tier 1 Post	Tier 2 Post	Tier 3 Recovery	Tier 4
Total portfolio	6	4	3	1	0
Per-direction (long or short)	4	3	2	1	0
Per-correlation-cluster	2	2	1	1	0
Per-strategy	3	2	2	1	0
Per-asset	1	1	1	1	0

Tier 4 is always 0 — no positions of any kind while an emergency shutdown is active. This cap is enforced independently by drawdown-kill-switch-trigger; this skill cross-checks it.

---

Correlation Clusters

Assets are grouped into correlation clusters based on empirically observed co-movement. A position in any asset within a cluster counts toward that cluster's cap. Clusters are not fixed — they are re-evaluated when the rolling 30-day correlation matrix changes significantly (threshold: correlation drift > 0.15 for a pair).

Default Cluster Definitions

CORRELATION_CLUSTERS = {
    "btc_majors": {
        "assets": ["BTC", "ETH", "SOL", "BNB"],
        "min_correlation": 0.70,   # observed 30d rolling
        "description": "Large-cap proof-of-work and L1 majors",
    },
    "eth_ecosystem": {
        "assets": ["ETH", "OP", "ARB", "MATIC", "BASE"],
        "min_correlation": 0.75,
        "description": "Ethereum and L2 ecosystem tokens",
    },
    "defi_bluechip": {
        "assets": ["AAVE", "UNI", "CRV", "SNX", "MKR"],
        "min_correlation": 0.65,
        "description": "DeFi protocol governance tokens",
    },
    "high_beta_alt": {
        "assets": ["DOGE", "SHIB", "PEPE", "WIF", "BONK"],
        "min_correlation": 0.60,
        "description": "High-beta meme and retail-driven tokens",
    },
    "perp_funding_plays": {
        "assets": [],   # populated dynamically from funding rate screening
        "min_correlation": ,
        : ,
    },
}




 () -> []:
     [
        name  name, cluster  CORRELATION_CLUSTERS.items()
         asset  cluster[]
    ]

---

Core Position Registry

from dataclasses import dataclass, field
from typing import Optional
import time

@dataclass
class OpenPosition:
    position_id: str          # unique identifier (uuid)
    asset: str                # e.g. "ETH"
    direction: str            # "long" or "short"
    strategy_id: str          # which strategy opened this position
    notional_usd: float       # current notional size
    margin_usd: float         # current margin consumed
    entry_timestamp_utc: float
    unrealised_pnl_usd: float = 0.0
    leverage: float = 1.0
    clusters: list[str] = field(default_factory=list)  # populated on creation

@dataclass
class PositionRegistry:
    positions: dict[str, OpenPosition] = field(default_factory=dict)
    # key: position_id

    def count_total(self) -> int:
        return len(self.positions)

    def count_by_direction(self, direction: ) -> :
         (  p  .positions.values()  p.direction == direction)

     () -> :
         (  p  .positions.values()  cluster  p.clusters)

     () -> :
         (  p  .positions.values()  p.strategy_id == strategy_id)

     () -> :
         (  p  .positions.values()  p.asset == asset)

     () -> :
         (p.margin_usd  p  .positions.values())

     () -> :
         (p.notional_usd  p  .positions.values())

     () -> :
        
        longs  = (p.notional_usd  p  .positions.values()  p.direction == )
        shorts = (p.notional_usd  p  .positions.values()  p.direction == )
         longs - shorts

---

Effective Cap Resolver

from enum import Enum

class KillSwitchTier(Enum):
    NONE   = 0
    TIER_1 = 1
    TIER_2 = 2
    TIER_3 = 3
    TIER_4 = 4

# Base caps — override via kill_switch_config.yaml
BASE_CAPS = {
    "total":       6,
    "directional": 4,
    "cluster":     2,
    "strategy":    3,
    "asset":       1,
}

# Tier overrides — each tier tightens caps relative to base
TIER_CAP_OVERRIDES = {
    KillSwitchTier.NONE:   {"total": 6,  "directional": 4, "cluster": 2, "strategy": 3, "asset": 1},
    KillSwitchTier.TIER_1: {"total": 4,  "directional": 3, "cluster": 2, "strategy": 2, "asset": 1},
    KillSwitchTier.TIER_2: {"total": 3,  "directional": 2, "cluster": 1, "strategy": 2, "asset": 1},
    KillSwitchTier.TIER_3: {: ,  : , : , : , : },
    KillSwitchTier.TIER_4: {: ,  : , : , : , : },
}

 () -> :
    
    caps = (TIER_CAP_OVERRIDES[kill_switch_tier])

    
     cascade_score >= :
        
        caps = {k: (caps[k], TIER_CAP_OVERRIDES[KillSwitchTier.TIER_3][k])
                 k  caps}
     cascade_score >= :
        
        caps[]   = (, caps[]   - )
        caps[] = (, caps[] - )

    
     funding_regime == :
        caps[]       = (, caps[]       - )
        caps[] = (, caps[] - )
     funding_regime == :
        caps[]       = (, caps[]       - )

    
     is_post_tier3_recovery:
        caps = {k: (caps[k], TIER_CAP_OVERRIDES[KillSwitchTier.TIER_3][k])
                 k  caps}

     caps

---

Admission Check

@dataclass
class AdmissionResult:
    admitted: bool
    binding_cap_level: str      # "total" | "directional" | "cluster" | "strategy" | "asset" | "none"
    binding_cap_value: int
    current_count_at_binding: int
    reason: str
    effective_caps: dict
    registry_snapshot: dict     # counts at time of check

def check_admission(
    registry: PositionRegistry,
    candidate_asset: str,
    candidate_direction: str,
    candidate_strategy_id: str,
    kill_switch_tier: KillSwitchTier,
    cascade_score: int,
    funding_regime: str,
    is_post_tier3_recovery: bool = False,
) -> AdmissionResult:
    """
    Check whether a new position may be opened.
    Returns AdmissionResult with admitted=True or False and full reasoning.
    """
    caps = resolve_effective_caps(
        kill_switch_tier, cascade_score, funding_regime, is_post_tier3_recovery
    )

    candidate_clusters = get_clusters_for_asset(candidate_asset)

    snapshot = {
        "total":       registry.count_total(),
        "directional": registry.count_by_direction(candidate_direction),
        "strategy":    registry.count_by_strategy(candidate_strategy_id),
        "asset":       registry.count_by_asset(candidate_asset),
        "clusters":    {c: registry.count_by_cluster(c) for c in candidate_clusters},
    }

    # Check each cap level in priority order (most restrictive first)

    # Level 5: per-asset (repeat position guard)
    if snapshot["asset"] >= caps["asset"]:
         AdmissionResult(
            admitted=,
            binding_cap_level=,
            binding_cap_value=caps[],
            current_count_at_binding=snapshot[],
            reason=
                   
                   ,
            effective_caps=caps,
            registry_snapshot=snapshot,
        )

    
     snapshot[] >= caps[]:
         AdmissionResult(
            admitted=,
            binding_cap_level=,
            binding_cap_value=caps[],
            current_count_at_binding=snapshot[],
            reason=
                   
                   ,
            effective_caps=caps,
            registry_snapshot=snapshot,
        )

    
     snapshot[] >= caps[]:
         AdmissionResult(
            admitted=,
            binding_cap_level=,
            binding_cap_value=caps[],
            current_count_at_binding=snapshot[],
            reason=
                   
                   
                   ,
            effective_caps=caps,
            registry_snapshot=snapshot,
        )

    
     cluster  candidate_clusters:
        cluster_count = snapshot[].get(cluster, )
         cluster_count >= caps[]:
             AdmissionResult(
                admitted=,
                binding_cap_level=,
                binding_cap_value=caps[],
                current_count_at_binding=cluster_count,
                reason=
                       
                       
                       ,
                effective_caps=caps,
                registry_snapshot=snapshot,
            )

    
     snapshot[] >= caps[]:
         AdmissionResult(
            admitted=,
            binding_cap_level=,
            binding_cap_value=caps[],
            current_count_at_binding=snapshot[],
            reason=
                   
                   ,
            effective_caps=caps,
            registry_snapshot=snapshot,
        )

    
     AdmissionResult(
        admitted=,
        binding_cap_level=,
        binding_cap_value=,
        current_count_at_binding=,
        reason=,
        effective_caps=caps,
        registry_snapshot=snapshot,
    )

---

Registry Maintenance

def register_position(
    registry: PositionRegistry,
    position: OpenPosition,
) -> PositionRegistry:
    """
    Add a newly opened position to the registry.
    Called immediately after confirmed exchange fill.
    Populates correlation clusters for the position.
    """
    position.clusters = get_clusters_for_asset(position.asset)
    registry.positions[position.position_id] = position
    return registry

def deregister_position(
    registry: PositionRegistry,
    position_id: str,
) -> PositionRegistry:
    """
    Remove a closed position from the registry.
    Called immediately after confirmed exchange close.
    """
    registry.positions.pop(position_id, None)
    return registry

def reconcile_registry(
    registry: PositionRegistry,
    exchange_positions: list[dict],  # from HyperLiquid /info clearinghouseState
) -> tuple[PositionRegistry, list[str]]:
    """
    Reconcile local registry against live exchange position state.
    Called at session open and after any API reconnect.
    Returns updated registry and list of discrepancy warnings.

    Discrepancy types:
    - Position in registry but not on exchange (phantom position)
    - Position on exchange but not in registry (shadow position)
    - Notional mismatch > 1% (partial fill or manual intervention)
    """
    warnings = []
    exchange_ids = {p["positionId"] for p in exchange_positions}
    registry_ids = set(registry.positions.keys())

    # Phantom positions
    for pid in registry_ids - exchange_ids:
        warnings.append(f"PHANTOM_POSITION: {pid} in registry but not on exchange. Removing.")
        registry.positions.pop(pid, None)

    
     ep  exchange_positions:
        pid = ep[]
         pid   registry_ids:
            warnings.append(
                
                
            )
            registry.positions[pid] = OpenPosition(
                position_id=pid,
                asset=ep[],
                direction=ep[],
                strategy_id=,
                notional_usd=ep[],
                margin_usd=ep[],
                entry_timestamp_utc=ep.get(, time.time()),
                leverage=ep.get(, ),
                clusters=get_clusters_for_asset(ep[]),
            )

    
     ep  exchange_positions:
        pid = ep[]
         pid  registry.positions:
            local_n  = registry.positions[pid].notional_usd
            remote_n = ep[]
             (local_n - remote_n) / (remote_n, ) > :
                warnings.append(
                    
                    
                )
                registry.positions[pid].notional_usd = remote_n

     registry, warnings

---

Regime-Driven Cap Tightening Logic

Beyond kill-switch tier, the following real-time signals automatically tighten caps within a session. The resolve_effective_caps() function handles these; this table documents the design intent:

Signal	Source Skill	Cap Tightening	Rationale
Cascade score ≥ 5 (HIGH)	liquidation-cascade-risk	total −1, cluster −1	Book fragility elevates correlated liquidation risk
Cascade score ≥ 8 (CRITICAL)	liquidation-cascade-risk	All caps → Tier 3 limits	Imminent cascade; portfolio must be minimal
Funding regime ELEVATED	high-funding-carry-avoidance	total −1	Carry cost silently reduces edge on all open positions
Funding regime EXTREME	high-funding-carry-avoidance	total −1, directional −1	Extreme funding: long positions bleed carry rapidly
Kill-switch Tier 1	drawdown-kill-switch-trigger	Per Tier 1 override table	Session soft stop in effect
Kill-switch Tier 3 recovery	drawdown-kill-switch-trigger	total=1 for first 5 sessions	Recovery protocol; avoid re-triggering Tier 3

---

Net Directional Exposure Check

Beyond position count, net directional notional must stay within bounds to prevent de facto one-sided book construction across multiple positions:

MAX_NET_DIRECTIONAL_PCT_NAV = 1.50   # 150% of NAV net long or net short

def check_net_directional_exposure(
    registry: PositionRegistry,
    candidate_direction: str,
    candidate_notional_usd: float,
    portfolio_nav_usd: float,
) -> dict:
    """
    Check whether adding the candidate notional would breach the net
    directional exposure cap. Separate from position count checks.
    """
    current_net = registry.net_direction_notional()
    candidate_signed = candidate_notional_usd if candidate_direction == "long" else -candidate_notional_usd
    projected_net = current_net + candidate_signed
    projected_net_pct_nav = abs(projected_net) / portfolio_nav_usd

    permitted = projected_net_pct_nav <= MAX_NET_DIRECTIONAL_PCT_NAV
    return {
        "permitted": permitted,
        "current_net_usd": current_net,
        "projected_net_usd": projected_net,
        "projected_net_pct_nav": projected_net_pct_nav * 100,
        "cap_pct_nav": MAX_NET_DIRECTIONAL_PCT_NAV * 100,
        "reason": (
            "Net directional exposure within cap."
            if permitted else
            f"Net directional cap breached: projected {projected_net_pct_nav*100:.1f}% "
            f"of NAV net {'long' if projected_net > 0 else 'short'} vs cap of "
            f"{MAX_NET_DIRECTIONAL_PCT_NAV*100:f}%. Reduce existing "
            
            
        ),
    }

---

Worked Example — Full Admission Pipeline

Inputs:
  Portfolio NAV:               $200,000
  Kill-switch tier:            NONE
  Cascade score:               5  (HIGH — tightens total -1, cluster -1)
  Funding regime:              ELEVATED (tightens total -1)
  Post-Tier-3 recovery:        False

Effective caps after regime modifiers:
  Base NONE caps:  total=6, directional=4, cluster=2, strategy=3, asset=1
  Cascade HIGH:    total=5, cluster=1
  Funding ELEVATED: total=4
  Final caps:      total=4, directional=4, cluster=1, strategy=3, asset=1

Current open positions:
  1. BTC-long  (btc_majors cluster, strategy: trending-bull)
  2. ETH-long  (btc_majors + eth_ecosystem clusters, strategy: trending-bull)
  3. OP-long   (eth_ecosystem cluster, strategy: funding-carry)

Candidate: SOL-long, strategy: trending-bull

Step 1 — Asset check:
  SOL count in registry: 0  <  asset cap 1  ✔

Step 2 — Total check:
  Total open: 3  <  effective total cap 4  ✔

Step 3 — Directional check:
  Long count: 3  <  directional cap 4  ✔

Step 4 — Cluster check:
  SOL clusters: [btc_majors]
  btc_majors count: 2 (BTC + ETH)  >=  cluster cap 1  ✗

Result: REJECTED
  binding_cap_level:  "cluster"
  reason: "Correlation cluster cap reached for 'btc_majors': 2/1
           positions in this cluster. SOL belongs to cluster
           'btc_majors'. Reduce correlated exposure before adding
           this position."

Implication: Agent should not add SOL while both BTC and ETH are
open. Options: close BTC or ETH first (freeing the btc_majors
cluster slot), or wait for a different asset in a different cluster.

---

Configuration

# multiclaw/metaclaw/concurrent_positions_config.yaml

concurrent_positions:
  base_caps:
    total:       6
    directional: 4
    cluster:     2
    strategy:    3
    asset:       1

  net_directional_cap_pct_nav: 150   # 150% of NAV net long or short

  correlation_clusters:
    btc_majors:
      assets: [BTC, ETH, SOL, BNB]
      min_correlation: 0.70
    eth_ecosystem:
      assets: [ETH, OP, ARB, MATIC]
      min_correlation: 0.75
    defi_bluechip:
      assets: [AAVE, UNI, CRV, SNX, MKR]
      min_correlation: 0.65
    high_beta_alt:
      assets: [DOGE, SHIB, PEPE, WIF, BONK]
      min_correlation: 0.60

  reconciliation:
    run_at_session_open: true
     
        
     

   
   

---

Audit JSONL Schema

{
  "event": "position_admission_rejected",
  "timestamp_utc": "2026-04-07T22:00:00Z",
  "candidate_asset": "SOL",
  "candidate_direction": "long",
  "candidate_strategy_id": "trending-bull",
  "candidate_notional_usd": 2353,
  "admitted": false,
  "binding_cap_level": "cluster",
  "binding_cap_value": 1,
  "current_count_at_binding": 2,
  "reason": "Correlation cluster cap reached for 'btc_majors': 2/1 positions.",
  "effective_caps": {"total": 4, "directional": 4      
   
   
   
   
     
     
     
     
      
  

---

Integration with Other Skills

• kelly-position-sizing-perps (risk/): Kelly produces final_notional_usd. This skill checks whether that position is even admissible before the notional is sent to execution. Kelly sizing and admission check are always run in sequence: Kelly first, admission second.
• drawdown-kill-switch-trigger (risk/): Tier state feeds directly into resolve_effective_caps(). On Tier 3 recovery, the maximum concurrent position count is 1 for the first 5 sessions — this skill enforces that constraint. On Tier 4, this skill cross- checks and returns admitted=False for all candidates.
• liquidation-cascade-risk (regime-detection/): Cascade score of 5+ reduces total and cluster caps within the session. Score of 8+ maps all caps to Tier 3 limits regardless of kill-switch state.
• high-funding-carry-avoidance (regime-detection/): Funding regime ELEVATED or EXTREME reduces total and directional caps. The carry skill's funding_regime field is a required input to resolve_effective_caps().
• slippage-budget-enforcement (execution/): Only called after this skill returns admitted=True. The execution pipeline order is: Kelly → Admission → Slippage → Order Submission.

---

Quick Decision Tree

New position candidate arrives — run in order:
│
├── 1. Check kill-switch state (drawdown-kill-switch-trigger)
│     state.trading_permitted()? → No → HALT. Do not proceed.
│
├── 2. Resolve effective caps
│     resolve_effective_caps(tier, cascade_score, funding_regime, recovery)
│
├── 3. Run admission check
│     result = check_admission(registry, asset, direction, strategy,
│                              tier, cascade_score, funding_regime)
│     result.admitted == False? → Log rejection. Do not proceed.
│     Inform strategy: reason + binding_cap_level
│
├── 4. Check net directional exposure
│     check_net_directional_exposure(registry, direction, notional, nav)
│     permitted == False? → Log rejection. Do not proceed.
│
├── 5. Proceed to slippage-budget-enforcement
│     Pass final_notional_usd as initial_size_usd
│
├── 6. On confirmed fill:
│     register_position(registry, new_position)
│     persist_state(registry)
│     Log position_admission_accepted audit event
│
└── 7. On confirmed close:
      deregister_position(registry, position_id)
      persist_state(registry)
      Log position_closed audit event