| """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()
|
