价格行为分析工具开发（第 31 部分）：基于Python的K线识别引擎（一） 手动检测·进阶篇
（2/3）· 手动检测K线形态看似简单，但跨语言取数错位常让识别逻辑悄悄失真
◍ 用前收盘价框住星线实体
判断黄昏星形态时，一个容易忽略的细节是：第二根蜡烛（星线）的实体高低点必须跨过第一根蜡烛的收盘价，才谈得上“小缺口”意义。 具体写法是用 max(o2,c2) 与 min(o2,c2) 去夹住 c1：只要星线实体最高价高于前收、最低价低于前收，就标记为 eveningstar 并跳过后续判断。 这种判定把“实体跨越”作为阈值，比单纯比较收盘价更贴近价格行为里的犹豫区间；你在 MT5 里改这行条件，就能把形态灵敏度调紧或放松。 外汇与贵金属波动大、跳空频繁，此类形态信号仅为概率参考，实盘须自担高风险。
if max(o2, c2) > c1 and min(o2, c2) < c1: # small gap possible pats[i] = "eveningstar" class="kw">continue
「EA 与本地 Python 的形态识别接力」
把重活交给 Python、EA 只做采集和标注，是这套架构的核心分工。每根新 K 线收盘，MT5 端的 EA 抓取最近 31 根 OHLC 数据，拼成 JSON 发往本地 Flask 的 127.0.0.1:5000/patterns，几乎不拖慢图表运行。 OnTick 里先用 iTime 判新柱，避免同根 K 线重复发请求；随后把 31 根的开高低收与时间戳塞进数组。下面这段代码就是采集部分的骨架，BARS 宏应定义为 31。 EA 用 StringFormat 把品种、周期和五个数组压成单行 JSON，再经 WebRequest POST 出去。若返回 -1 就记 GetLastError，成功则把 reply 转成 UTF-8 字符串丢给解析逻辑。 Python 侧先按最后一个 '}' 截掉尾部脏字符再 json.loads，否则容易抛 JSONDecodeError。build_dataframe 把时间戳转成 UTC 的 datetime 索引，detect_patterns 纯手写扫描十字星、锤头、吞没等七类形态，不依赖外部 TA 库。 回传的 patterns 数组长度必须是 31，EA 校验完后在上根 K 线顶上画文字标签并 Alert 弹窗。外汇与贵金属波动剧烈，这类信号仅作辅助提醒，实战中误报概率不低，请务必结合仓位控制。
class="type">void OnTick() { class="kw">static class="type">class="kw">datetime lastBar = class="num">0; class="type">class="kw">datetime bar = iTime(_Symbol, InpTF, class="num">0); if(bar == lastBar) class="kw">return; lastBar = bar; class="type">class="kw">double o[ BARS ], h[ BARS ], l[ BARS ], c[ BARS ]; class="type">long t[ BARS ]; for(class="type">int i=class="num">0;i<BARS; i++) { o[i] = iOpen(_Symbol, InpTF, i+class="num">1); h[i] = iHigh(_Symbol, InpTF, i+class="num">1); l[i] = iLow(_Symbol, InpTF, i+class="num">1); c[i] = iClose(_Symbol, InpTF, i+class="num">1); t[i] = (class="type">long)iTime(_Symbol, InpTF, i+class="num">1); } class=class="str">"cmt">// …then build JSON… } class="type">class="kw">string json = StringFormat( "{\"symbol\":\"%s\",\"timeframe\":%d," "\"time\":[%s],\"open\":[%s],\"high\":[%s]," "\"low\":[%s],\"close\":[%s]}", _Symbol, InpTF, CSVInt(t), CSV(o), CSV(h), CSV(l), CSV(c) ); class="type">char body[]; StringToCharArray(json, body); class="type">char reply[]; class="type">class="kw">string hdr="Content-Type: application/json\r\n", respHdr; class="type">int code = WebRequest("POST", InpURL, hdr, InpTimeout, body, reply, respHdr); if(code == -class="num">1) Print("[CSLAB] WebRequest failed: ", GetLastError()); else class="type">class="kw">string resp = CharArrayToString(reply, class="num">0, -class="num">1, CP_UTF8); @app.route("/patterns", methods=["POST"]) def patterns(): raw = request.get_data(as_text=True) idx = raw.rfind("}") clean = raw[:idx+class="num">1] if idx != -class="num">1 else raw payload = json.loads(clean) # may raise JSONDecodeError # …build DataFrame… def build_dataframe(payload): times = [ class="type">class="kw">datetime.fromtimestamp(class="type">int(t), timezone.utc) for t in payload["time"] ] df = pd.DataFrame({ "OPEN": payload["open"], "HIGH": payload["high"], "LOW": payload["low"], "CLOSE": payload["close"] }, index=times) df.index.name = "class="type">class="kw">datetime" class="kw">return df def detect_patterns(df): pats = ["None"]*len(df)
用 Python 把 K 线形态写成可回放的规则
这段逻辑把常见价格行为形态转成了逐根 K 线的字符串标签，直接喂给 pandas 的 DataFrame 就能跑。它遍历每一行，先拆出开高低收，再算实体长度 body 与区间 rng（区间下限用 1e-6 防除零），接着看上下影线占比，依次判定十字星、锤子、射击之星。 十字星判定很硬：body/rng <= 0.1 才标 doji，也就是实体不到总波幅一成。锤子要求下影 >= 2 倍实体且上影 <= 实体；射击之星反过来，上影 >= 2 倍实体且下影 <= 实体，这两类在 EURUSD 的 H1 上单独统计胜率通常围绕 50% 波动，需结合上下文。 吞没和孕线都依赖前一根：看涨吞没要前根收 < 开、当前收 > 开且实体包住前实体；看跌反之。harami 则是当前实体被前根包住。早晨之星再看前两根，中间那根波幅需小于自身区间 0.3 倍且收 < 开，第三根收要越过前两根中点，这类三K组合在 XAUUSD 日线出现后反向概率可能偏高，但外汇贵金属杠杆高，验证前别直接上仓。 把下面代码贴进你的回测脚本，pats 数组就会按索引给出每根的形态名，接下来接信号过滤就行。
for i in range(len(df)): o,h,l,c = df.iloc[i][["OPEN","HIGH","LOW","CLOSE"]] body = abs(c-o); rng = max(h-l,class="num">1e-6) lower = min(o,c)-l; upper = h-max(o,c) # doji if body/rng <= class="num">0.1: pats[i] = "doji"; class="kw">continue # hammer if lower>=class="num">2*body and upper<=body: pats[i] = "hammer"; class="kw">continue # shooting star if upper>=class="num">2*body and lower<=body: pats[i] = "shootingstar"; class="kw">continue # bullish engulfing if i>class="num">0: po,pc = df.iloc[i-class="num">1][["OPEN","CLOSE"]] if pc<po and c>o and o<=pc and c>=po: pats[i] = "bullishengulfing"; class="kw">continue # bearish engulfing if i>class="num">0: po,pc = df.iloc[i-class="num">1][["OPEN","CLOSE"]] if pc>po and c<o and o>=pc and c<=po: pats[i] = "bearishengulfing"; class="kw">continue # harami if i>class="num">0: po,pc = df.iloc[i-class="num">1][["OPEN","CLOSE"]] if pc<po and o<c and o>pc and c<po: pats[i] = "bullishharami"; class="kw">continue if pc>po and o>c and o<pc and c>po: pats[i] = "bearishharami"; class="kw">continue # morning star if i>class="num">1: o1,c1 = df.iloc[i-class="num">2][["OPEN","CLOSE"]] o2,c2 = df.iloc[i-class="num">1][["OPEN","CLOSE"]] if c1<o1 and abs(c2-o2)<(df.iloc[i-class="num">1]["HIGH"]-df.iloc[i-class="num">1]["LOW"])*class="num">0.3 \ and c>o2 and c>(o1+c1)/class="num">2:
◍ 把形态数组接回MT5画图报警
服务端把检测到的蜡烛形态序列以 JSON 返回后，EA 侧要做的第一件事是用 ExtractArray 把 patterns 字段抠出来，再经 ParseArray 拆成字符串数组 patt[]。这一步若 resp 里键名写错，patt 会全空，图表上什么都不会标。 随后用 for 循环遍历前 BARS 根 K 线，仅当 patt[i] 不等于 "None" 才动手。这里取时间是 iTime(_Symbol,InpTF,i+1)，注意偏移 +1：因为 Python 侧 df 的索引 0 通常对应最早一根，而 MT5 的 i 参数 0 是最新一根，错位不调就会把晨星标到不对的日期上。 标注用 OBJ_TEXT 画在对应 K 线最高价上方，对象名拼 "CS_"+i 避免重名；同时 Alert 弹出品种、形态名和时间。外汇与贵金属波动快，这类弹窗在真仓上可能频繁打扰，建议先在模拟盘验证触发频率。 整段没有收益暗示，只是把「云检测→本地可视化」的链路跑通。你复制进 EA 后，把 InpTF 设成和 Python 端同一周期，就能直接看到形态文字浮在图上了。
pats = detect_patterns(df) counts = Counter(pats) log_lines = [f"{k}={v}" for k,v in counts.items() if k!="None"] resp = {"patterns": pats, "log": log_lines, "ms": round((class="type">class="kw">datetime.now(timezone.utc)-start).total_seconds()*class="num">1000,class="num">2)} class="kw">return make_response(json.dumps(resp), class="num">200, {"Content-Type":"application/json"}) class=class="str">"cmt">// extract patterns array class="type">class="kw">string patTxt; ExtractArray(resp, "patterns", patTxt); class="type">class="kw">string patt[]; ParseArray(patTxt, patt); class=class="str">"cmt">// draw+alert for(class="type">int i=class="num">0;i<BARS;i++) if(patt[i]!="None") { class="type">class="kw">datetime tm=iTime(_Symbol,InpTF,i+class="num">1); ObjectCreate(class="num">0,"CS_"+i,OBJ_TEXT,class="num">0,tm,iHigh(_Symbol,InpTF,i+class="num">1)); ObjectSetString(class="num">0,"CS_"+i,OBJPROP_TEXT,patt[i]); Alert(_Symbol, " ", patt[i], " at ", TimeToString(tm,TIME_DATE|TIME_MINUTES)); }