python-mt5-support-resistance-indicator/python/zone_reference.py

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