Portfolio Heat: Aggregate Open Risk and Correlated Gaps

from math import isfinite

# id, side, quantity, entry, mark, stop, point_value, USD_per_settlement, exit_cost_USD
positions = [
    ("A", 1, 100, 100, 105, 95, 1, 1, 10),
    ("B", 1, 200, 50, 52, 48, 1, 1, 10),
    ("Future", -1, 1, 5000, 4990, 5010, 50, 1, 10),
    ("C", 1, 100, 80, 82, 76, 1, 1.10, 10),
]

def finite_number(x):
    return type(x) in (int, float) and isfinite(x)

def heat(rows, equity, *, snapshot_complete=False):
    if snapshot_complete is not True:
        raise ValueError("Incomplete position/stop/FX snapshot")
    if not finite_number(equity) or equity <= 0:
        raise ValueError("Equity must be finite and positive")
    entry_total, mark_total, seen = 0.0, 0.0, set()
    detail = []
    for row in rows:
        if len(row) != 9:
            raise ValueError("Expected all nine position fields")
        name, side, qty, entry, mark, stop, point, fx, cost = row
        if not isinstance(name, str) or not name or name in seen:
            raise ValueError("Missing or duplicate position id")
        seen.add(name)
        if not all(finite_number(x) for x in row[1:]):
            raise ValueError(name + ": missing/non-finite numeric input")
        if side not in (-1, 1) or min(qty, point, fx) <= 0 or cost < 0:
            raise ValueError(name + ": invalid side, size, units or cost")
        if side * (mark - stop) <= 0:
            raise ValueError(name + ": stop reached or on wrong side")
        scale = qty * point * fx
        er = max(0, side * (entry - stop)) * scale + cost
        mr = side * (mark - stop) * scale + cost
        if not all(isfinite(x) for x in (scale, er, mr)):
            raise ValueError(name + ": numeric overflow")
        detail.append((name, er, mr))
        entry_total += er
        mark_total += mr
    eh, mh = 100 * entry_total / equity, 100 * mark_total / equity
    if not all(isfinite(x) for x in (entry_total, mark_total, eh, mh)):
        raise ValueError("Total or ratio overflow")
    return detail, entry_total, mark_total, eh, mh

result = heat(positions, 100000, snapshot_complete=True)
for name, er, mr in result[0]:
    print(name, round(er, 2), round(mr, 2))
print(tuple(round(x, 2) for x in result[1:]))
# (1880.0, 3500.0, 1.88, 3.5)

# Joint basis/FX stress: assumed exits, not guaranteed fills.
stress_exits = [90, 44, 5015, 68]
stress_fx = [1, 1, 1, 1.20]
losses = [
    side * (mark - fill) * qty * point * rate + cost
    for (_, side, qty, entry, mark, stop, point, fx, cost), fill, rate
    in zip(positions, stress_exits, stress_fx)
]
# C is a fully paid EUR cash equity. Revalue its entire holding in USD.
# Current value: 82 * 100 * 1.10; exit proceeds: 68 * 100 * 1.20.
losses[3] = 82 * 100 * 1.10 - 68 * 100 * 1.20 + 10
print([round(x, 2) for x in losses], round(sum(losses), 2))
# [1510, 1610, 1260, 870.0], 5250.0