#!/usr/bin/env python3 """Assumed trading-bot capital and cost arithmetic; Python 3.10+, stdlib only. No prices or venue rules are fetched, no orders are placed, and no strategy return, sustainable income or probability of a loss sequence is estimated. """ from __future__ import annotations import argparse import csv import hashlib import json from decimal import Decimal, ROUND_CEILING, ROUND_FLOOR, getcontext from pathlib import Path getcontext().prec = 40 def decimal(value) -> Decimal: result = value if isinstance(value, Decimal) else Decimal(str(value)) if not result.is_finite(): raise ValueError("All monetary values, returns and constraints must be finite") return result def positive(value, name: str) -> Decimal: result = decimal(value) if result <= 0: raise ValueError(f"{name} must be positive") return result def nonnegative(value, name: str) -> Decimal: result = decimal(value) if result < 0: raise ValueError(f"{name} must be nonnegative") return result def sizing_parameters(settings: dict) -> dict: price = positive(settings["reference_price"], "reference price") increment = positive(settings["quantity_increment"], "quantity increment") minimum = positive(settings["minimum_reference_notional"], "minimum notional") risk_fraction = positive(settings["planned_risk_fraction"], "risk fraction") stop = positive(settings["stop_distance_fraction"], "stop distance") entry_fee = nonnegative(settings["entry_fee_fraction"], "entry fee") exit_fee = nonnegative(settings["exit_fee_fraction"], "exit fee") allowance = nonnegative(settings["adverse_execution_allowance_each_side"], "execution allowance") friction = nonnegative(settings["linear_round_trip_friction_fraction"], "linear friction") if risk_fraction > 1 or stop >= 1 or allowance >= 1: raise ValueError("Risk fraction must be at most one; stop and execution allowance must be below one") if friction != entry_fee + exit_fee + 2 * allowance: raise ValueError("Linear friction must equal both fees plus both adverse execution allowances") return {"price": price, "increment": increment, "minimum": minimum, "risk_fraction": risk_fraction, "stop": stop, "entry_fee": entry_fee, "exit_fee": exit_fee, "allowance": allowance, "friction": friction, "entry_cash_factor": (1 + allowance) * (1 + entry_fee), "planned_loss_fraction_of_notional": stop + friction} def floor_to_increment(quantity: Decimal, increment: Decimal) -> Decimal: return (quantity / increment).to_integral_value(rounding=ROUND_FLOOR) * increment def minimum_executable_capital(settings: dict) -> dict: p = sizing_parameters(settings) minimum_quantity = (p["minimum"] / (p["price"] * p["increment"])).to_integral_value(rounding=ROUND_CEILING) * p["increment"] executable_notional = minimum_quantity * p["price"] risk_minimum = executable_notional * p["planned_loss_fraction_of_notional"] / p["risk_fraction"] cash_minimum = executable_notional * p["entry_cash_factor"] exit_price = p["price"] * (1 - p["stop"]) * (1 - p["allowance"]) minimum_same_quantity_for_exit = (p["minimum"] / (exit_price * p["increment"])).to_integral_value(rounding=ROUND_CEILING) * p["increment"] entry_and_stop_quantity = max(minimum_quantity, minimum_same_quantity_for_exit) entry_and_stop_notional = entry_and_stop_quantity * p["price"] entry_and_stop_risk_minimum = entry_and_stop_notional * p["planned_loss_fraction_of_notional"] / p["risk_fraction"] entry_and_stop_cash_minimum = entry_and_stop_notional * p["entry_cash_factor"] return { "assumed_reference_price": p["price"], "assumed_quantity_increment": p["increment"], "assumed_minimum_reference_notional": p["minimum"], "minimum_executable_quantity": minimum_quantity, "minimum_executable_reference_notional": executable_notional, "risk_budget_minimum_equity": risk_minimum, "entry_cash_minimum_equity": cash_minimum, "minimum_equity_satisfying_both": max(risk_minimum, cash_minimum), "binding_constraint": "risk_budget" if risk_minimum >= cash_minimum else "entry_cash", "constraint_scope": "entry-only minimum, conservative planned loss budget, and funded entry cash", "assumed_full_quantity_stop_exit_minimum_diagnostic": { "assumed_adversely_adjusted_stop_exit_price_before_fee": exit_price, "minimum_entry_quantity_also_meeting_stated_stop_exit_minimum": entry_and_stop_quantity, "corresponding_entry_reference_notional": entry_and_stop_notional, "corresponding_stop_exit_notional_before_exit_fee": entry_and_stop_quantity * exit_price, "risk_budget_minimum_equity": entry_and_stop_risk_minimum, "entry_cash_minimum_equity": entry_and_stop_cash_minimum, "minimum_equity_for_entry_and_assumed_stop_exit_minimum": max(entry_and_stop_risk_minimum, entry_and_stop_cash_minimum), "meaning": "Additional assumed same-quantity notional check; not a full order-lifecycle or guaranteed-exit model", }, } def size_position(equity, settings: dict) -> dict: equity = positive(equity, "equity") p = sizing_parameters(settings) risk_budget = equity * p["risk_fraction"] risk_notional = risk_budget / p["planned_loss_fraction_of_notional"] cash_notional = equity / p["entry_cash_factor"] unrounded_notional = min(risk_notional, cash_notional) candidate_quantity = floor_to_increment(unrounded_notional / p["price"], p["increment"]) candidate_notional = candidate_quantity * p["price"] executable = candidate_quantity > 0 and candidate_notional >= p["minimum"] quantity = candidate_quantity if executable else Decimal(0) notional = quantity * p["price"] entry_cash = notional * p["entry_cash_factor"] planned_loss = notional * p["planned_loss_fraction_of_notional"] # Stop executes exactly at the assumed reference stop less the allowance. # This is a model identity, not a promise that a stop actually fills there. assumed_stop_exit_proceeds = notional * (1 - p["stop"]) * (1 - p["allowance"]) * (1 - p["exit_fee"]) modeled_gap_free_stop_loss = entry_cash - assumed_stop_exit_proceeds stop_exit_notional = notional * (1 - p["stop"]) * (1 - p["allowance"]) return { "equity": equity, "risk_budget": risk_budget, "risk_fraction": p["risk_fraction"], "risk_budget_notional_before_rounding": risk_notional, "cash_affordable_notional_before_rounding": cash_notional, "candidate_quantity_rounded_down": candidate_quantity, "candidate_reference_notional": candidate_notional, "status": "EXECUTE_IN_MODEL" if executable else "SKIP_BELOW_ASSUMED_MINIMUM", "quantity": quantity, "reference_notional": notional, "reference_notional_over_equity": notional / equity, "entry_cash_required": entry_cash, "cash_remaining_after_entry": equity - entry_cash, "conservative_planned_gap_free_loss": planned_loss, "planned_loss_over_equity": planned_loss / equity, "unused_risk_budget": risk_budget - planned_loss, "modeled_gap_free_stop_loss_with_multiplicative_costs": modeled_gap_free_stop_loss, "assumed_stopped_exit_notional_before_exit_fee": stop_exit_notional, "meets_assumed_same_minimum_on_stopped_exit": executable and stop_exit_notional >= p["minimum"], } def monthly_break_even(deployed_fraction, round_trips, gross_return, friction, fixed_cost, equity=None) -> dict: allocation = positive(deployed_fraction, "deployed fraction") count = positive(round_trips, "round trips") gross = decimal(gross_return) cost = nonnegative(friction, "friction") fixed = nonnegative(fixed_cost, "fixed cost") if allocation > 1 or count != count.to_integral_value(): raise ValueError("Unleveraged deployed fraction must be at most one; round trips must be an integer") turnover = count * allocation surplus_rate = turnover * (gross - cost) if surplus_rate > 0: threshold = fixed / surplus_rate status = "POSITIVE_SURPLUS_FIXED_COST_THRESHOLD" if fixed > 0 else "NO_FIXED_COST_THRESHOLD_TRADING_CONSTRAINTS_REMAIN" elif surplus_rate == 0 and fixed == 0: threshold = None status = "ALL_POSITIVE_EQUITIES_BREAK_EVEN_NO_OPERATING_SURPLUS" else: threshold = None status = "NO_POSITIVE_EQUITY_BREAK_EVEN" result = { "deployed_fraction": allocation, "round_trips_per_month": int(count), "assumed_gross_notional_return_per_round_trip": gross, "assumed_linear_round_trip_friction": cost, "fixed_monthly_cost": fixed, "round_trip_reference_notional_sum_over_equity": turnover, "linear_monthly_surplus_rate_before_fixed_cost": surplus_rate, "break_even_equity": threshold, "status": status, } if equity is not None: equity = positive(equity, "equity") result.update({ "equity": equity, "round_trip_reference_notional_sum": turnover * equity, "assumed_gross_monthly_pnl": turnover * equity * gross, "assumed_variable_monthly_friction": turnover * equity * cost, "linear_monthly_pnl_before_fixed_cost": surplus_rate * equity, "linear_monthly_pnl_after_fixed_cost": surplus_rate * equity - fixed, "linear_monthly_return_after_fixed_cost": surplus_rate - fixed / equity, "fixed_monthly_cost_over_equity": fixed / equity, "required_gross_notional_return_per_round_trip_for_break_even": cost + fixed / (turnover * equity), }) return result def income_equity(desired_income, fixed_cost, assumed_monthly_return) -> dict: income = nonnegative(desired_income, "desired income") fixed = nonnegative(fixed_cost, "fixed cost") rate = positive(assumed_monthly_return, "assumed monthly return") return { "desired_monthly_income_after_fixed_cost_before_tax": income, "assumed_monthly_return_after_variable_cost_before_fixed_cost": rate, "fixed_monthly_cost": fixed, "arithmetically_required_equity": (income + fixed) / rate, } def consecutive_losses(starting_equity, loss_fraction, count: int, fixed_dollar_loss) -> dict: initial = positive(starting_equity, "starting equity") fraction = positive(loss_fraction, "loss fraction") fixed = nonnegative(fixed_dollar_loss, "fixed dollar loss") if fraction >= 1 or not isinstance(count, int) or count < 0: raise ValueError("Loss fraction must be below one; event count must be a nonnegative integer") remaining = initial * (1 - fraction) ** count fixed_remaining = initial - fixed * count if fixed_remaining <= 0: raise ValueError("Fixed-dollar comparator exhausts the account; this narrow comparison requires positive equity") return { "starting_equity": initial, "assumed_fractional_loss_each_event": fraction, "consecutive_loss_events": count, "fixed_fraction_remaining_equity": remaining, "fixed_fraction_drawdown": 1 - remaining / initial, "fixed_fraction_recovery_return_required": initial / remaining - 1, "fixed_dollar_loss_each_event": fixed, "fixed_dollar_remaining_equity": fixed_remaining, "fixed_dollar_drawdown": 1 - fixed_remaining / initial, "fixed_dollar_recovery_return_required": initial / fixed_remaining - 1, } def serializable(value): if isinstance(value, Decimal): return float(value) if isinstance(value, dict): return {key: serializable(item) for key, item in value.items()} if isinstance(value, (list, tuple)): return [serializable(item) for item in value] return value def calculate(protocol: dict) -> dict: sizing = protocol["position_sizing"] grid_sizing = {**sizing, **sizing["coarse_grid_example"]} b = protocol["break_even"] allocation = b["fraction_of_reference_equity_deployed_per_round_trip"] count = b["round_trips_per_month"] gross = b["gross_notional_return_per_round_trip_main_case"] friction = b["linear_variable_friction_per_round_trip"] income = protocol["income_arithmetic"] losses = protocol["consecutive_loss_arithmetic"] return { "notice": protocol["purpose"], "assumption_provenance": protocol["assumption_provenance"], "currency": protocol["currency"], "position_sizing": [size_position(equity, sizing) for equity in protocol["equities"]], "minimum_executable_capital": { "main_grid": minimum_executable_capital(sizing), "coarse_grid": minimum_executable_capital(grid_sizing), }, "monthly_cost_impact": [ {"scenario_family": "capital_and_fixed_cost", **monthly_break_even(allocation, count, gross, friction, fixed, equity)} for equity in protocol["equities"] for fixed in b["fixed_monthly_cost_tiers"] ], "gross_return_sensitivity": [ {"scenario_family": "gross_return_and_fixed_cost", **monthly_break_even(allocation, count, assumed_gross, friction, fixed)} for assumed_gross in b["gross_notional_return_per_round_trip_scenarios"] for fixed in b["fixed_monthly_cost_tiers"] ], "friction_sensitivity": [ {"scenario_family": "friction_at_main_gross_return", **monthly_break_even(allocation, count, gross, assumed_friction, b["main_fixed_monthly_cost"])} for assumed_friction in b["friction_sensitivity_at_main_gross_return"] ], "income_arithmetic": [income_equity(income["desired_monthly_income_after_fixed_cost_before_tax"], income["fixed_monthly_cost"], rate) for rate in income["assumed_monthly_account_returns_after_variable_costs_before_fixed_cost"]], "consecutive_losses": [consecutive_losses(losses["starting_equity"], losses["loss_fraction_of_current_equity_each_event"], count, losses["fixed_dollar_loss_comparator"]) for count in losses["event_counts"]], "limitations": protocol["excluded"], } def write_csv(path: Path, records: list[dict]) -> None: columns = list(dict.fromkeys(key for record in records for key in record)) with path.open("w", encoding="utf-8", newline="") as stream: writer = csv.DictWriter(stream, fieldnames=columns) writer.writeheader() writer.writerows(records) def run(protocol_path: Path, output_dir: Path) -> dict: protocol = json.loads(protocol_path.read_text(encoding="utf-8"), parse_float=Decimal) results = calculate(protocol) results["protocol_sha256"] = hashlib.sha256(protocol_path.read_bytes()).hexdigest() output_dir.mkdir(parents=True, exist_ok=True) (output_dir / "results.json").write_text(json.dumps(serializable(results), indent=2, allow_nan=False) + "\n", encoding="utf-8") write_csv(output_dir / "position-sizing.csv", results["position_sizing"]) write_csv(output_dir / "break-even.csv", results["monthly_cost_impact"] + results["gross_return_sensitivity"] + results["friction_sensitivity"]) write_csv(output_dir / "consecutive-losses.csv", results["consecutive_losses"]) return results if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--protocol", type=Path, default=Path(__file__).resolve().parent / "protocol.json") parser.add_argument("--output-dir", type=Path, default=Path(__file__).resolve().parent / "results") arguments = parser.parse_args() run(arguments.protocol, arguments.output_dir) print("Calculated assumed capital, sizing and cost examples. No market data, network requests or live orders.")