KRWUSD model

KRWUSD-model.py

"""
https://www.reddit.com/r/quant/comments/1s5woy4/krwusd_i_think_korean_fx_peg_is_doubtful_so_i/
"""

from dataclasses import dataclass
from enum import Enum
from math import exp
from typing import Dict, Optional


class MarketState(str, Enum):
    NORMAL = "normal"
    ALERT = "alert"
    PANIC_BUYING = "panic_buying"
    SQUEEZE = "squeeze"
    TRAP_FAILURE = "trap_failure"


@dataclass
class RegimeParams:
    """
    Approximate reconstruction from the handwritten state table.
    lambda_decay: how quickly intervention impact fades with time
    k: sensitivity / steepness near important levels
    """
    lambda_decay: float
    k: float


REGIMES: Dict[MarketState, RegimeParams] = {
    MarketState.NORMAL: RegimeParams(lambda_decay=0.45, k=35.0),
    MarketState.ALERT: RegimeParams(lambda_decay=0.275, k=12.5),
    MarketState.PANIC_BUYING: RegimeParams(lambda_decay=0.15, k=0.3),
    MarketState.SQUEEZE: RegimeParams(lambda_decay=1.0, k=1_000_000.0),   # proxy for "∞"
    MarketState.TRAP_FAILURE: RegimeParams(lambda_decay=0.0001, k=0.0001),  # proxy for "0+"
}


@dataclass
class FXInterventionInputs:
    """
    Reconstructed variables from the page.
    """
    # Time
    t: float                  # current time
    t_max: float              # time of peak intervention / key event

    # Price / exchange rate
    e_current: float          # current exchange rate E
    e_anchor: float           # anchor / defended / reference exchange rate C
    delta_e_exp: float        # expected exchange-rate deviation / spread

    # Volumes / liquidity
    v_base: float             # base liquidity
    v_intervention: float     # direct intervention size / flow
    v_macro: float = 0.0      # optional macro / broad liquidity
    v_micro: float = 0.0      # optional microstructure liquidity

    # Alpha terms
    alpha_base: float = 1.0
    alpha_jaw: float = 0.0
    alpha_dep: float = 1.0

    # Stability / band
    epsilon: float = 1.0
    lower_bound: Optional[float] = None
    upper_bound: Optional[float] = None

    # State
    state: MarketState = MarketState.NORMAL


class FXInterventionFlowModel:
    """
    Practical Python reconstruction of the handwritten model.
    This is not an academically validated model; it is a structured
    implementation of the apparent reasoning on the page.
    """

    def __init__(self, regimes: Dict[MarketState, RegimeParams] = REGIMES):
        self.regimes = regimes

    @staticmethod
    def adjusted_alpha(alpha_base: float, alpha_jaw: float) -> float:
        """
        Recreates the line:
            alpha_adj = alpha_base + delta_alpha_jaw
        """
        return alpha_base + alpha_jaw

    @staticmethod
    def time_decay(lambda_decay: float, t: float, t_max: float) -> float:
        """
        Recreates the exponential fading:
            exp(-lambda * (t - T_max))
        """
        dt = max(t - t_max, 0.0)
        return exp(-lambda_decay * dt)

    @staticmethod
    def level_sensitivity(e_current: float, e_anchor: float, epsilon: float) -> float:
        """
        Recreates the denominator logic:
            1 / ((C - E)^2 + epsilon)

        Stronger response when current rate is close to defended / key level.
        """
        return 1.0 / (((e_anchor - e_current) ** 2) + epsilon)

    @staticmethod
    def in_psychological_band(
        e_current: float,
        lower_bound: Optional[float],
        upper_bound: Optional[float]
    ) -> bool:
        if lower_bound is None or upper_bound is None:
            return False
        lo = min(lower_bound, upper_bound)
        hi = max(lower_bound, upper_bound)
        return lo <= e_current <= hi

    def net_response(self, x: FXInterventionInputs) -> float:
        """
        Reconstructed from the middle formula:
            V_net = V_base * exp(-lambda*(t-Tmax)) * 1/((C-E)^2 + epsilon)

        We also allow regime k to scale response.
        """
        regime = self.regimes[x.state]
        decay = self.time_decay(regime.lambda_decay, x.t, x.t_max)
        sensitivity = self.level_sensitivity(x.e_current, x.e_anchor, x.epsilon)

        # k is treated as regime sensitivity multiplier
        return x.v_base * decay * sensitivity * regime.k

    def gross_intervention_effect(self, x: FXInterventionInputs) -> float:
        """
        Reconstructed from the top line conceptually:
            gross effect = adjusted intervention flow + expectation term

        This is an interpretation, because the exact handwritten formula
        is partially obscured.
        """
        alpha_adj = self.adjusted_alpha(x.alpha_base, x.alpha_jaw)

        # flow contribution
        flow_term = alpha_adj * x.v_intervention

        # expectation / depreciation / spread term
        # use max(t, small value) to avoid division by zero
        expectation_term = x.alpha_dep * (x.delta_e_exp / max(x.t, 1e-9))

        # optional liquidity context
        liquidity_term = x.v_macro + x.v_micro

        return flow_term + expectation_term + liquidity_term

    def band_multiplier(self, x: FXInterventionInputs) -> float:
        """
        Psychological threshold logic:
        inside band -> more attention / friction
        outside band -> breakout risk / regime transition

        This is an implementation choice, not directly visible as a formula.
        """
        if x.lower_bound is None or x.upper_bound is None:
            return 1.0

        if self.in_psychological_band(x.e_current, x.lower_bound, x.upper_bound):
            return 1.25   # more sensitivity inside watched zone
        return 0.9        # less friction once away from the zone

    def total_score(self, x: FXInterventionInputs) -> float:
        """
        Final combined score.
        Positive means stronger intervention support / stabilization force.
        """
        gross = self.gross_intervention_effect(x)
        net = self.net_response(x)
        band = self.band_multiplier(x)

        return gross + (net * band)

    def explain(self, x: FXInterventionInputs) -> Dict[str, float]:
        regime = self.regimes[x.state]
        alpha_adj = self.adjusted_alpha(x.alpha_base, x.alpha_jaw)
        decay = self.time_decay(regime.lambda_decay, x.t, x.t_max)
        sensitivity = self.level_sensitivity(x.e_current, x.e_anchor, x.epsilon)
        net = self.net_response(x)
        gross = self.gross_intervention_effect(x)
        band = self.band_multiplier(x)
        total = self.total_score(x)

        return {
            "alpha_adj": alpha_adj,
            "lambda_decay": regime.lambda_decay,
            "k": regime.k,
            "time_decay": decay,
            "level_sensitivity": sensitivity,
            "gross_effect": gross,
            "net_response": net,
            "band_multiplier": band,
            "total_score": total,
        }


if __name__ == "__main__":
    model = FXInterventionFlowModel()

    # Example based loosely on the handwritten notes
    inputs = FXInterventionInputs(
        t=14.0,
        t_max=13.5,
        e_current=1524.0,
        e_anchor=1520.0,
        delta_e_exp=6.0,
        v_base=100.0,
        v_intervention=80.0,
        v_macro=10.0,
        v_micro=5.0,
        alpha_base=1.0,
        alpha_jaw=0.3,
        alpha_dep=0.8,
        epsilon=25.0 / 12.0,
        lower_bound=1510.0,
        upper_bound=1535.0,
        state=MarketState.ALERT,
    )

    result = model.explain(inputs)

    print("FX Intervention Flow Model")
    print("-" * 32)
    for key, value in result.items():
        print(f"{key:20s}: {value:.6f}")