python-mt5-support-resistance-indicator/tests/verify.py

"""Synthetic verification. Native MetaEditor compilation is a separate check."""
import csv
import importlib.util
import math
from pathlib import Path
import shutil
import subprocess
import tempfile
from datetime import datetime, timedelta

ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location('zones', ROOT/'python/zone_reference.py')
zones = importlib.util.module_from_spec(spec)
spec.loader.exec_module(zones)


def bars(n=240):
    result = []
    start = datetime(2025, 1, 1)
    for i in range(n):
        price = 2000+5*math.sin(i/6)+math.sin(i/2)
        result.append(dict(time=(start+timedelta(minutes=i)).isoformat(' '),
                           open=price-.1, close=price, high=price+.7, low=price-.8))
    return result


def body(source, name):
    start = source.index(('bool ' if name == 'SRPivot' else 'double ' if name == 'SRHalfWidth' else 'int ')+name+'(')
    opening = source.index('{', start)
    depth, i = 1, opening+1
    while depth:
        depth += (source[i] == '{')-(source[i] == '}')
        i += 1
    return source[start:i]


def main():
    sample = bars()
    expected = zones.detect(sample, show_broken=True, max_zones=100)
    assert expected
    for z in expected:
        assert z['available'] > z['pivot']
        assert z['lower'] < z['upper']
        assert not z['broken'] or z['ended'] > z['available']
    forming_changed = [dict(b) for b in sample]
    forming_changed[-1].update(high=999999, low=.01, close=123)
    assert zones.detect(forming_changed, show_broken=True, max_zones=100) == expected
    small = bars(6)
    for b in small:
        b.update(open=10, close=10, high=11, low=9)
    small[2]['low'] = 8
    assert zones.detect(small[:-1], period=1) == []
    one = zones.detect(small, period=1)
    assert len(one) == 1 and one[0]['pivot'] == small[2]['time']
    assert one[0]['available'] == small[5]['time']
    tied = [dict(b) for b in small]
    tied[1]['low'] = 8
    assert zones.detect(tied, period=1) == []
    assert len(zones.detect(sample, max_zones=3, show_broken=True)) == 3
    with tempfile.TemporaryDirectory() as directory:
        directory = Path(directory)
        f = directory/'bars.csv'
        with f.open('w', newline='') as out:
            writer = csv.writer(out, delimiter='\t')
            writer.writerow(['<DATE>','<TIME>','<OPEN>','<HIGH>','<LOW>','<CLOSE>'])
            for b in sample:
                date, time = b['time'].split()
                writer.writerow([date.replace('-', '.'),time,b['open'],b['high'],b['low'],b['close']])
        assert zones.load_bars(f) == sample
        print('Python checks passed: confirmation delay, forming exclusion, strict ties, bounds, cap and CSV import.')
        if not shutil.which('g++'):
            print('C++ parity checks SKIPPED: install g++ to run them; native MT5 compilation still required.')
            return
        source = (ROOT/'MQL5/Indicators/SR_Research_Zones.mq5').read_text()
        core = '\n'.join(body(source, n) for n in ('SRPivot','SRHalfWidth','SRBreakIndex'))
        core = core.replace('const MqlRates &r[]', 'const std::vector<MqlRates> &r')
        cpp = '''#include <vector>
#include <algorithm>
#include <cmath>
#include <iostream>
#include <iomanip>
#include <cassert>
struct MqlRates { double open, high, low, close; };
double MathAbs(double x){return std::abs(x);}
double MathMax(double a,double b){return std::max(a,b);}
'''+core+'\nint main(){std::vector<MqlRates> r={\n'
        cpp += ',\n'.join('{'+','.join(repr(b[k]) for k in ('open','high','low','close'))+'}' for b in sample)
        cpp += '''};
std::cout<<std::setprecision(17);
for(int i=12;i+2<=(int)r.size()-2;i++)for(int side:{1,-1}){
 if(!SRPivot(r,i,2,side))continue;
 double a=side>0?r[i].low:r[i].high;
 double w=SRHalfWidth(r,i+2,14,.25,.01);
 int b=SRBreakIndex(r,i+3,r.size()-2,side,a-w,a+w,2);
 std::cout<<i<<","<<side<<","<<a-w<<","<<a+w<<","<<b<<"\\n";
}
std::vector<MqlRates> q(5,{10,11,9,10});
q[0].close=8;q[1].close=10;q[2].close=8;q[3].close=8;
assert(SRBreakIndex(q,0,3,1,9,11,2)==3);
q[2].close=9;assert(SRBreakIndex(q,0,3,1,9,11,2)==-1);
q[0].close=12;q[1].close=12;
assert(SRBreakIndex(q,0,1,-1,9,11,2)==1);
}
'''
        (directory/'test.cpp').write_text(cpp)
        binary = directory/'test'
        subprocess.run(['g++','-std=c++17','-O2',str(directory/'test.cpp'),'-o',str(binary)], check=True)
        result = subprocess.run([str(binary)], capture_output=True, text=True, check=True)
        actual = list(csv.reader(result.stdout.splitlines()))
        assert len(actual) == len(expected)
        for row, z in zip(actual, expected):
            i, side, lo, hi, broken = row
            assert sample[int(i)]['time'] == z['pivot']
            assert ('SUPPORT' if int(side)>0 else 'RESISTANCE') == z['side']
            assert math.isclose(float(lo), z['lower'], abs_tol=1e-10)
            assert math.isclose(float(hi), z['upper'], abs_tol=1e-10)
            assert (int(broken)>=0) == z['broken']
            if z['broken']:
                assert sample[int(broken)+1]['time'] == z['ended']
        print(f'C++ parity passed: {len(actual)} zone records; actual MQL pivot, width and break bodies.')
        print('Break checks passed: consecutive closes, reset at boundary and sell-side symmetry.')


if __name__ == '__main__':
    main()