125 lines
6.5 KiB
Python
125 lines
6.5 KiB
Python
import unittest
|
|
|
|
import pandas as pd
|
|
|
|
from primethai.backtest import run_backtest
|
|
from primethai.config import RegimeHeat, StrategyConfig, ThaiMarketRules, TradingCosts
|
|
from primethai.portfolio import HeatPosition, portfolio_cushion, portfolio_heat, round_quantity, round_to_tick, size_position
|
|
|
|
|
|
def feature_frame():
|
|
idx = pd.bdate_range("2024-01-02", periods=5)
|
|
return pd.DataFrame({
|
|
"open": [9.8, 9.9, 10.5, 9.9, 9.3], "high": [10.1, 10.2, 13.1, 10.2, 9.7],
|
|
"low": [9.6, 9.8, 10.4, 9.8, 9.0], "close": [9.9, 10.2, 12.8, 9.4, 9.2],
|
|
"volume": [1_000_000] * 5, "avg_value_turnover20": [50_000_000] * 5,
|
|
"ema21_high": [10.5] * 5, "ema21_low": [9.5] * 5, "ema21_close": [10.0] * 5,
|
|
"atr14": [1.0] * 5, "extension_atr": [.2] * 5, "trend_state": ["UP"] * 5,
|
|
"price_state": ["INSIDE_STRUCTURE"] * 5, "rs_percentile": [95] * 5,
|
|
"regime": ["CONFIRMED_UPTREND"] * 5, "new_risk_allowed": [True] * 5,
|
|
"mco_z": [0.0] * 5, "mcsi_state": ["UP"] * 5, "weekly_trend_state": ["UP"] * 5,
|
|
"higher_low": [False] * 5, "volume_contracting": [True] * 5,
|
|
}, index=idx)
|
|
|
|
|
|
def market_frame(index):
|
|
return pd.DataFrame({"regime": ["CONFIRMED_UPTREND"] * len(index),
|
|
"new_risk_allowed": [True] * len(index)}, index=index)
|
|
|
|
|
|
class PortfolioBacktestTests(unittest.TestCase):
|
|
def test_position_sizing_respects_risk_and_position_cap(self):
|
|
sized = size_position(1_000_000, .005, 20, 18, lot_size=100, max_position_pct=.20)
|
|
self.assertEqual(sized["shares"], 2500)
|
|
self.assertEqual(sized["initial_risk"], 5000)
|
|
self.assertEqual(sized["position_value"], 50_000)
|
|
capped = size_position(1_000_000, .005, 20, 19.9, lot_size=100, max_position_pct=.02)
|
|
self.assertEqual(capped["shares"], 1000)
|
|
self.assertEqual(capped["position_value"], 20_000)
|
|
|
|
def test_thai_board_lot_rounding_and_tick_grid(self):
|
|
self.assertEqual(round_quantity(199, 100), 100)
|
|
self.assertEqual(round_quantity(199, 50), 150)
|
|
self.assertEqual(round_to_tick(25.24, ThaiMarketRules(), direction="down"), 25.0)
|
|
self.assertEqual(round_to_tick(25.24, ThaiMarketRules(), direction="up"), 25.25)
|
|
self.assertEqual(round_to_tick(10.17, ThaiMarketRules(), direction="down"), 10.1)
|
|
self.assertEqual(round_to_tick(400.0, ThaiMarketRules()), 400.0)
|
|
|
|
def test_portfolio_heat_and_cushion(self):
|
|
positions = [HeatPosition(initial_risk=1000, market_value=11000, cost_basis=10000),
|
|
HeatPosition(initial_risk=500, market_value=4500, cost_basis=5000)]
|
|
self.assertEqual(portfolio_heat(positions, 100_000), .015)
|
|
self.assertEqual(portfolio_cushion(positions, 100_000), .005)
|
|
|
|
def _run(self, frame, *, market=None, **kwargs):
|
|
strategy = StrategyConfig(initial_equity=100_000, weakness_risk_pct=.01,
|
|
confirmation_risk_pct=.01, max_position_pct=.5,
|
|
min_value_turnover_thb=0, min_price=0,
|
|
enable_weekly=False, heat=RegimeHeat(confirmed_uptrend=1.0))
|
|
return run_backtest({"ADVANC": frame}, market if market is not None else market_frame(frame.index), strategy=strategy,
|
|
rules=ThaiMarketRules(), costs=TradingCosts(commission_rate=0, slippage_bps=0), **kwargs)
|
|
|
|
def test_market_regime_from_market_engine_blocks_new_risk(self):
|
|
frame = feature_frame()
|
|
blocked = market_frame(frame.index)
|
|
blocked["regime"] = "CORRECTION"
|
|
blocked["new_risk_allowed"] = False
|
|
result = self._run(frame, market=blocked)
|
|
self.assertTrue(result.fills.empty)
|
|
self.assertTrue(result.trades.empty)
|
|
|
|
def test_accurate_membership_gates_new_entries_on_the_signal_date(self):
|
|
frame = feature_frame()
|
|
market = market_frame(frame.index)
|
|
signal_dates = {frame.index[1], frame.index[2]}
|
|
market.attrs["members_by_date"] = {
|
|
pd.Timestamp(date): ([] if date in signal_dates else ["ADVANC"])
|
|
for date in frame.index
|
|
}
|
|
|
|
result = self._run(frame, market=market)
|
|
|
|
self.assertTrue(result.fills.empty)
|
|
self.assertTrue(result.trades.empty)
|
|
|
|
def test_next_open_execution_two_r_partial_and_runner_attribution(self):
|
|
frame = feature_frame()
|
|
result = self._run(frame)
|
|
self.assertFalse(result.fills.empty)
|
|
buy = result.fills[result.fills["side"] == "BUY"].iloc[0]
|
|
self.assertEqual(buy["date"], frame.index[2])
|
|
self.assertEqual(buy["price"], frame.loc[frame.index[2], "open"])
|
|
trade = result.trades.iloc[0]
|
|
self.assertTrue(trade["two_r_hit"])
|
|
self.assertGreater(trade["trim_2r_pnl"], 0)
|
|
self.assertLess(trade["runner_pnl"], 0)
|
|
self.assertAlmostEqual(trade["total_pnl"], trade["trim_2r_pnl"] + trade["runner_pnl"] + trade["dividend_cash"])
|
|
trim = result.fills[result.fills["reason"] == "2R_trim"].iloc[0]
|
|
self.assertGreater(trim["shares"], 0)
|
|
self.assertLess(trim["shares"], buy["shares"])
|
|
exit_fill = result.fills[result.fills["reason"] == "close_below_ema21_low_next_open"].iloc[0]
|
|
self.assertEqual(exit_fill["date"], frame.index[4])
|
|
|
|
def test_dividends_are_credited_while_position_is_open(self):
|
|
frame = feature_frame()
|
|
frame["dividends"] = [0, 0, 0, .5, 0]
|
|
frame["stock_splits"] = [0, 0, 0, 0, 0]
|
|
result = self._run(frame)
|
|
self.assertGreater(result.trades.iloc[0]["dividend_cash"], 0)
|
|
|
|
def test_unadjusted_split_option_adjusts_open_runner_shares(self):
|
|
frame = feature_frame()
|
|
price_fields = ["open", "high", "low", "close", "ema21_high", "ema21_low", "ema21_close"]
|
|
frame.loc[frame.index[3], price_fields] /= 2
|
|
frame.loc[frame.index[3], "stock_splits"] = 2.0
|
|
frame.loc[frame.index[4], price_fields] /= 2
|
|
frame["stock_splits"] = frame["stock_splits"].fillna(0)
|
|
result = self._run(frame, yahoo_split_adjusted=False)
|
|
buy = result.fills[result.fills["side"] == "BUY"].iloc[0]
|
|
trim = result.fills[result.fills["reason"] == "2R_trim"].iloc[0]
|
|
runner_exit = result.fills[result.fills["reason"] == "close_below_ema21_low_next_open"].iloc[0]
|
|
self.assertEqual(result.trades.iloc[0]["initial_shares"], buy["shares"])
|
|
self.assertEqual(runner_exit["shares"], (buy["shares"] - trim["shares"]) * 2)
|
|
self.assertAlmostEqual(result.trades.iloc[0]["total_pnl"],
|
|
result.trades.iloc[0]["trim_2r_pnl"] + result.trades.iloc[0]["runner_pnl"])
|