MetaTrader5 MCP Server

import unittest from unittest.mock import MagicMock, patch import pandas as pd import numpy as np import sys import os # Add src to path sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../src'))) from mtdata.forecast.barriers import forecast_barrier_hit_probabilities, forecast_barrier_closed_form, forecast_barrier_optimize class TestForecastBarriers(unittest.TestCase): def setUp(self): self.mock_mt5 = MagicMock() self.mock_mt5.symbol_info.return_value = MagicMock(digits=5, point=0.00001) # Patch mt5 in barriers module self.mt5_patcher = patch('mtdata.forecast.barriers.mt5', self.mock_mt5) self.mt5_patcher.start() # Mock fetch_history self.fetch_history_patcher = patch('mtdata.forecast.barriers._fetch_history') self.mock_fetch_history = self.fetch_history_patcher.start() # Create dummy dataframe dates = pd.date_range(start='2023-01-01', periods=500, freq='H') prices = np.linspace(1.0, 1.1, 500) + np.random.normal(0, 0.001, 500) self.df = pd.DataFrame({'time': dates, 'close': prices}) self.mock_fetch_history.return_value = self.df def _set_flat_history(self, price: float = 1.0, bars: int = 200): dates = pd.date_range(start='2023-01-01', periods=bars, freq='h') closes = np.full(bars, float(price)) self.mock_fetch_history.return_value = pd.DataFrame({'time': dates, 'close': closes}) def tearDown(self): self.mt5_patcher.stop() self.fetch_history_patcher.stop() def test_forecast_barrier_hit_probabilities(self): result = forecast_barrier_hit_probabilities( symbol="EURUSD", timeframe="H1", horizon=10, method="mc_gbm", direction="long", tp_pct=0.5, sl_pct=0.5 ) self.assertIn("success", result) self.assertTrue(result["success"]) self.assertIn("prob_tp_first", result) self.assertIn("prob_sl_first", result) def test_forecast_barrier_bootstrap(self): result = forecast_barrier_hit_probabilities( symbol="EURUSD", timeframe="H1", horizon=10, method="bootstrap", direction="long", tp_pct=0.5, sl_pct=0.5, params={"block_size": 5} ) self.assertIn("success", result) self.assertTrue(result["success"]) self.assertEqual(result["method"], "bootstrap") def test_forecast_barrier_garch(self): try: import arch except ImportError: self.skipTest("arch package not installed") result = forecast_barrier_hit_probabilities( symbol="EURUSD", timeframe="H1", horizon=10, method="garch", direction="long", tp_pct=0.5, sl_pct=0.5 ) self.assertIn("success", result) self.assertTrue(result["success"]) self.assertEqual(result["method"], "garch") def test_forecast_barrier_closed_form(self): result = forecast_barrier_closed_form( symbol="EURUSD", timeframe="H1", horizon=10, direction="up", barrier=1.2 ) self.assertIn("success", result) self.assertTrue(result["success"]) self.assertIn("prob_hit", result) def test_forecast_barrier_optimize(self): result = forecast_barrier_optimize( symbol="EURUSD", timeframe="H1", horizon=10, method="mc_gbm", direction="long", mode="pct", tp_min=0.1, tp_max=0.5, tp_steps=3, sl_min=0.1, sl_max=0.5, sl_steps=3, objective="edge" ) self.assertIn("success", result) self.assertTrue(result["success"]) self.assertIn("best", result) self.assertIn("grid", result) def test_forecast_barrier_optimize_refine_and_metrics(self): # Deterministic paths to verify refine pass and ranking self._set_flat_history(1.0) paths = np.array([ [1.0, 1.01, 1.02, 1.03], [1.0, 0.99, 0.98, 0.97], [1.0, 1.002, 0.998, 1.006], ]) with patch('mtdata.forecast.barriers._simulate_gbm_mc') as mock_sim: mock_sim.return_value = {"price_paths": paths} result = forecast_barrier_optimize( symbol="EURUSD", timeframe="H1", horizon=4, method="mc_gbm", direction="long", mode="pct", tp_min=0.5, tp_max=1.0, tp_steps=2, sl_min=0.5, sl_max=1.0, sl_steps=2, objective="kelly_uncond", refine=True, refine_radius=0.4, refine_steps=3, return_grid=True, ) self.assertTrue(result["success"]) grid = result.get("grid") self.assertTrue(grid) # Base grid has 4 combos; refine should add more unique candidates self.assertGreater(len(grid), 4) self.assertIn("prob_no_hit", grid[0]) self.assertIn("t_hit_tp_median", grid[0]) kelly_vals = [g["kelly_uncond"] for g in grid] self.assertEqual(kelly_vals, sorted(kelly_vals, reverse=True)) def test_forecast_barrier_optimize_summary_truncates_grid(self): self._set_flat_history(1.0) paths = np.array([ [1.0, 1.01, 1.02, 1.03], [1.0, 0.99, 0.98, 0.97], [1.0, 1.002, 0.998, 1.006], ]) with patch('mtdata.forecast.barriers._simulate_gbm_mc') as mock_sim: mock_sim.return_value = {"price_paths": paths} result = forecast_barrier_optimize( symbol="EURUSD", timeframe="H1", horizon=4, method="mc_gbm", direction="long", mode="pct", tp_min=0.5, tp_max=1.0, tp_steps=2, sl_min=0.5, sl_max=1.0, sl_steps=2, objective="edge", refine=False, output="summary", top_k=2, return_grid=True, ) self.assertTrue(result["success"]) grid = result.get("grid") self.assertTrue(grid) self.assertEqual(len(grid), 2) self.assertEqual(result["best"], grid[0]) if __name__ == '__main__': unittest.main()