| """OHLC reference for SR_Research_Zones.mq5. Standard-library Python only.
|
| Input: MT5 chart-bar CSV (comma, semicolon or tab), not a raw tick export.
|
| The last input bar is treated as forming and excluded from calculations.
|
| """
|
| import argparse
|
| import csv
|
| import math
|
| from datetime import datetime
|
|
|
|
|
| def load_bars(path):
|
| with open(path, newline='', encoding='utf-8-sig') as f:
|
| sample = f.read(8192)
|
| f.seek(0)
|
| dialect = csv.Sniffer().sniff(sample, delimiters=',;\t')
|
| reader = csv.DictReader(f, dialect=dialect)
|
| bars = []
|
| for row in reader:
|
| row = {k.strip().strip('<>').lower(): v for k, v in row.items()}
|
| dt = datetime.fromisoformat(row['date'].replace('.', '-') + ' ' + row['time'])
|
| b = dict(time=dt.isoformat(' '), **{k: float(row[k]) for k in ('open','high','low','close')})
|
| if not all(math.isfinite(b[k]) for k in ('open','high','low','close')):
|
| raise ValueError('Nonfinite price')
|
| if min(b[k] for k in ('open','high','low','close')) <= 0 or not (
|
| b['low'] <= min(b['open'], b['close']) <= max(b['open'], b['close']) <= b['high']):
|
| raise ValueError('Invalid OHLC candle')
|
| if bars and b['time'] <= bars[-1]['time']:
|
| raise ValueError('Bars must have unique timestamps in ascending order')
|
| bars.append(b)
|
| return bars
|
|
|
|
|
| def detect(bars, point=.001, history=2000, wing=2, period=14,
|
| multiplier=.25, minimum_points=10, break_closes=2,
|
| invalidate=True, show_broken=False, max_zones=20):
|
| if not all(math.isfinite(x) for x in (point, multiplier, minimum_points)):
|
| raise ValueError('Parameters must be finite')
|
| if point <= 0 or history < 100 or not 1 <= wing <= 20 or period < 1:
|
| raise ValueError('Invalid point size, history, swing or ATR period')
|
| if multiplier <= 0 or minimum_points <= 0 or break_closes < 1 or max_zones < 1:
|
| raise ValueError('Invalid width, break count or zone count')
|
| r = bars[-history:]
|
| last = len(r)-2
|
| zones = []
|
| for i in range(max(wing, period-wing), last-wing+1):
|
| c = i+wing
|
| for side in (1, -1):
|
| field = 'low' if side > 0 else 'high'
|
| anchor = r[i][field]
|
| neighbors = [r[j][field] for j in range(i-wing, i+wing+1) if j != i]
|
| if not all(anchor < x if side > 0 else anchor > x for x in neighbors):
|
| continue
|
| tr = [max(r[j]['high']-r[j]['low'], abs(r[j]['high']-r[j-1]['close']),
|
| abs(r[j]['low']-r[j-1]['close'])) for j in range(c-period+1, c+1)]
|
| width = max(minimum_points*point, multiplier*sum(tr)/period)
|
| lower, upper = anchor-width, anchor+width
|
| broken, run = None, 0
|
| if invalidate:
|
| for j in range(c+1, last+1):
|
| outside = r[j]['close'] < lower if side > 0 else r[j]['close'] > upper
|
| run = run+1 if outside else 0
|
| if run >= break_closes:
|
| broken = r[j+1]['time']
|
| break
|
| if broken and not show_broken:
|
| continue
|
| zones.append(dict(side='SUPPORT' if side > 0 else 'RESISTANCE',
|
| pivot=r[i]['time'], available=r[c+1]['time'],
|
| lower=lower, upper=upper, broken=bool(broken), ended=broken or ''))
|
| return zones[-max_zones:]
|
|
|
|
|
| def main():
|
| p = argparse.ArgumentParser(description=__doc__)
|
| p.add_argument('bars_csv')
|
| p.add_argument('--point', type=float, required=True, help='MT5 SYMBOL_POINT for this symbol')
|
| p.add_argument('--out', default='zones.csv')
|
| p.add_argument('--swing', type=int, default=2)
|
| p.add_argument('--history', type=int, default=2000)
|
| p.add_argument('--show-broken', action='store_true')
|
| args = p.parse_args()
|
| zones = detect(load_bars(args.bars_csv), point=args.point, wing=args.swing,
|
| history=args.history, show_broken=args.show_broken)
|
| with open(args.out, 'w', newline='', encoding='utf-8') as f:
|
| w = csv.DictWriter(f, fieldnames=['side','pivot','available','lower','upper','broken','ended'])
|
| w.writeheader()
|
| w.writerows(zones)
|
| print(f'{len(zones)} zones written to {args.out}; final bar excluded as forming.')
|
|
|
|
|
| if __name__ == '__main__':
|
| main()
|
