265 lines
11 KiB
MQL5
265 lines
11 KiB
MQL5
//+------------------------------------------------------------------+
|
|
//| DataHarvester.mqh |
|
|
//| Centaur Quant Architecture — Data Module |
|
|
//| Lightweight Live Market Telemetry Harvester |
|
|
//+------------------------------------------------------------------+
|
|
//| PURPOSE |
|
|
//| Captures Bid/Ask/Spread/TickVolume on a controlled cadence and |
|
|
//| packages them as SDP Tick_Harvest frames via CSDPEncoder. Emits |
|
|
//| periodic SDP Heartbeat frames. Zero blocking, O(1) per pass — |
|
|
//| safe to call from OnTick/OnTimer. Dispatch is decoupled: the |
|
|
//| caller drains pending frames via TakePending() and forwards them |
|
|
//| through the transport (CSocketClient). |
|
|
//+------------------------------------------------------------------+
|
|
#property strict
|
|
|
|
#ifndef DATAHARVESTER_MQH
|
|
#define DATAHARVESTER_MQH
|
|
|
|
#include "../Network/CSDPEncoder.mqh"
|
|
|
|
//+------------------------------------------------------------------+
|
|
//| |
|
|
//+------------------------------------------------------------------+
|
|
class CDataHarvester
|
|
{
|
|
private:
|
|
string m_symbol; // instrument being harvested
|
|
string m_timeframe; // SDP wire timeframe (e.g., "M15")
|
|
CSDPEncoder m_encoder; // SDP serializer — single source of truth
|
|
|
|
ulong m_harvest_interval_ms; // min gap between Tick_Harvest encodes
|
|
ulong m_heartbeat_interval_ms; // period of Heartbeat emits
|
|
ulong m_last_harvest_ms; // GetTickCount64() of last harvest
|
|
ulong m_last_heartbeat_ms; // GetTickCount64() of last heartbeat
|
|
bool m_stale_logged; // one-shot stale-tick warning flag
|
|
|
|
bool m_ready; // symbol validated
|
|
|
|
//--- pending outbound queue (decoupled from transport) ---
|
|
string m_queue[];
|
|
int m_queue_head;
|
|
enum { MAX_PENDING = 64, COMPACT_THRESHOLD = 512 };
|
|
|
|
//--- telemetry counters ---
|
|
ulong m_ticks_harvested;
|
|
ulong m_heartbeats_emitted;
|
|
|
|
string PeriodName(const ENUM_TIMEFRAMES period);
|
|
void Enqueue(const string payload);
|
|
void CompactQueue();
|
|
|
|
public:
|
|
CDataHarvester(const string symbol, const string timeframe = "");
|
|
~CDataHarvester();
|
|
|
|
//--- one lightweight pass; call from OnTick or OnTimer ---
|
|
bool Harvest();
|
|
|
|
//--- drain pending SDP payloads (caller forwards via CSocketClient) ---
|
|
bool HasPending() const;
|
|
bool TakePending(string &out_payload);
|
|
int PendingCount() const;
|
|
|
|
//--- runtime configuration (clamped to sane minimums) ---
|
|
void SetHarvestInterval(const ulong interval_ms);
|
|
void SetHeartbeatInterval(const ulong interval_ms);
|
|
|
|
//--- read access ---
|
|
bool IsReady() const { return m_ready; }
|
|
string Symbol() const { return m_symbol; }
|
|
string Timeframe() const { return m_timeframe; }
|
|
ulong TicksHarvested() const { return m_ticks_harvested; }
|
|
ulong HeartbeatsEmitted() const { return m_heartbeats_emitted; }
|
|
};
|
|
//+------------------------------------------------------------------+
|
|
//| Constructor — bind symbol/timeframe, validate, reset timers. |
|
|
//+------------------------------------------------------------------+
|
|
CDataHarvester::CDataHarvester(const string symbol, const string timeframe)
|
|
: m_symbol(symbol),
|
|
m_timeframe(timeframe),
|
|
m_harvest_interval_ms(1000),
|
|
m_heartbeat_interval_ms(30000),
|
|
m_last_harvest_ms(0),
|
|
m_last_heartbeat_ms(0),
|
|
m_stale_logged(false),
|
|
m_ready(false),
|
|
m_queue_head(0),
|
|
m_ticks_harvested(0),
|
|
m_heartbeats_emitted(0)
|
|
{
|
|
if(StringLen(m_symbol) == 0)
|
|
{
|
|
PrintFormat("[CDataHarvester] ERROR: empty symbol in constructor.");
|
|
return;
|
|
}
|
|
if(StringLen(m_timeframe) == 0)
|
|
m_timeframe = PeriodName((ENUM_TIMEFRAMES)_Period);
|
|
// light validation: a live instrument exposes a positive point value
|
|
m_ready = (SymbolInfoDouble(m_symbol, SYMBOL_POINT) > 0.0);
|
|
if(!m_ready)
|
|
PrintFormat("[CDataHarvester] ERROR: symbol '%s' not available in the terminal.", m_symbol);
|
|
else
|
|
PrintFormat("[CDataHarvester] INFO: ready on %s %s (harvest %I64u ms, heartbeat %I64u ms).",
|
|
m_symbol, m_timeframe, m_harvest_interval_ms, m_heartbeat_interval_ms);
|
|
}
|
|
//+------------------------------------------------------------------+
|
|
//| Destructor — nothing to release; present for symmetry. |
|
|
//+------------------------------------------------------------------+
|
|
CDataHarvester::~CDataHarvester()
|
|
{
|
|
}
|
|
//+------------------------------------------------------------------+
|
|
//| Harvest — single non-blocking pass. |
|
|
//| 1) Emits a Heartbeat frame when the heartbeat interval elapsed. |
|
|
//| 2) Emits a Tick_Harvest frame when the harvest interval elapsed |
|
|
//| AND a fresh tick is available (stale ticks are skipped). |
|
|
//| Returns true when at least one frame was queued. |
|
|
//+------------------------------------------------------------------+
|
|
bool CDataHarvester::Harvest()
|
|
{
|
|
if(!m_ready)
|
|
return false;
|
|
const ulong now = GetTickCount64();
|
|
bool emitted = false;
|
|
//--- 1) periodic heartbeat ---
|
|
if(m_last_heartbeat_ms == 0 || (now - m_last_heartbeat_ms) >= m_heartbeat_interval_ms)
|
|
{
|
|
Enqueue(m_encoder.EncodeHeartbeat(m_symbol, m_timeframe));
|
|
m_last_heartbeat_ms = now;
|
|
m_heartbeats_emitted++;
|
|
emitted = true;
|
|
}
|
|
//--- 2) throttled tick harvest ---
|
|
if(m_last_harvest_ms == 0 || (now - m_last_harvest_ms) >= m_harvest_interval_ms)
|
|
{
|
|
MqlTick tick;
|
|
if(!SymbolInfoTick(m_symbol, tick) || tick.bid <= 0.0 || tick.ask <= 0.0)
|
|
{
|
|
if(!m_stale_logged)
|
|
{
|
|
PrintFormat("[CDataHarvester] WARNING: no valid tick for %s. Harvest skipped.", m_symbol);
|
|
m_stale_logged = true;
|
|
}
|
|
return emitted;
|
|
}
|
|
// fresh-tick gate: skip when the market is closed (stale last tick)
|
|
if(tick.time <= 0 || (TimeCurrent() - tick.time) > 60)
|
|
{
|
|
if(!m_stale_logged)
|
|
{
|
|
PrintFormat("[CDataHarvester] WARNING: market closed or tick stale for %s. Harvest skipped.", m_symbol);
|
|
m_stale_logged = true;
|
|
}
|
|
return emitted;
|
|
}
|
|
m_stale_logged = false;
|
|
const string payload = m_encoder.EncodeTickHarvest(m_symbol, m_timeframe,
|
|
tick.bid, tick.ask,
|
|
tick.time, (long)tick.volume);
|
|
if(StringLen(payload) > 0)
|
|
{
|
|
Enqueue(payload);
|
|
m_ticks_harvested++;
|
|
emitted = true;
|
|
}
|
|
m_last_harvest_ms = now;
|
|
}
|
|
return emitted;
|
|
}
|
|
//+------------------------------------------------------------------+
|
|
//| Enqueue — append a payload to the pending queue. |
|
|
//| When full, the OLDEST pending frame is dropped (ring semantics). |
|
|
//+------------------------------------------------------------------+
|
|
void CDataHarvester::Enqueue(const string payload)
|
|
{
|
|
if(StringLen(payload) == 0)
|
|
return;
|
|
const int size = ArraySize(m_queue);
|
|
if(size - m_queue_head >= MAX_PENDING)
|
|
{
|
|
PrintFormat("[CDataHarvester] WARNING: pending queue full (%d). Oldest payload dropped.",
|
|
MAX_PENDING);
|
|
m_queue_head++;
|
|
return;
|
|
}
|
|
ArrayResize(m_queue, size + 1);
|
|
m_queue[size] = payload;
|
|
}
|
|
//+------------------------------------------------------------------+
|
|
//| HasPending — true when at least one frame awaits dispatch. |
|
|
//+------------------------------------------------------------------+
|
|
bool CDataHarvester::HasPending() const
|
|
{
|
|
return (ArraySize(m_queue) - m_queue_head) > 0;
|
|
}
|
|
//+------------------------------------------------------------------+
|
|
//| TakePending — pop the oldest pending frame; false when empty. |
|
|
//| Compacted lazily to bound memory. |
|
|
//+------------------------------------------------------------------+
|
|
bool CDataHarvester::TakePending(string &out_payload)
|
|
{
|
|
const int size = ArraySize(m_queue);
|
|
if(m_queue_head >= size)
|
|
{
|
|
m_queue_head = 0;
|
|
ArrayResize(m_queue, 0);
|
|
return false;
|
|
}
|
|
out_payload = m_queue[m_queue_head++];
|
|
if(m_queue_head >= COMPACT_THRESHOLD)
|
|
CompactQueue();
|
|
return true;
|
|
}
|
|
//+------------------------------------------------------------------+
|
|
//| PendingCount — frames awaiting dispatch. |
|
|
//+------------------------------------------------------------------+
|
|
int CDataHarvester::PendingCount() const
|
|
{
|
|
return ArraySize(m_queue) - m_queue_head;
|
|
}
|
|
//+------------------------------------------------------------------+
|
|
//| CompactQueue — shift live frames to the front and shrink. |
|
|
//+------------------------------------------------------------------+
|
|
void CDataHarvester::CompactQueue()
|
|
{
|
|
const int size = ArraySize(m_queue);
|
|
if(m_queue_head <= 0)
|
|
return;
|
|
const int remaining = size - m_queue_head;
|
|
string tmp[];
|
|
if(remaining > 0)
|
|
ArrayCopy(tmp, m_queue, 0, m_queue_head, remaining);
|
|
ArrayResize(m_queue, remaining);
|
|
if(remaining > 0)
|
|
ArrayCopy(m_queue, tmp);
|
|
m_queue_head = 0;
|
|
}
|
|
//+------------------------------------------------------------------+
|
|
//| SetHarvestInterval — throttle for Tick_Harvest frames. |
|
|
//+------------------------------------------------------------------+
|
|
void CDataHarvester::SetHarvestInterval(const ulong interval_ms)
|
|
{
|
|
m_harvest_interval_ms = MathMax(interval_ms, (ulong)100);
|
|
}
|
|
//+------------------------------------------------------------------+
|
|
//| SetHeartbeatInterval — period for Heartbeat frames. |
|
|
//+------------------------------------------------------------------+
|
|
void CDataHarvester::SetHeartbeatInterval(const ulong interval_ms)
|
|
{
|
|
m_heartbeat_interval_ms = MathMax(interval_ms, (ulong)1000);
|
|
}
|
|
//+------------------------------------------------------------------+
|
|
//| PeriodName — wire name of an ENUM_TIMEFRAMES value. |
|
|
//+------------------------------------------------------------------+
|
|
string CDataHarvester::PeriodName(const ENUM_TIMEFRAMES period)
|
|
{
|
|
ENUM_TIMEFRAMES p = period;
|
|
if(p == PERIOD_CURRENT)
|
|
p = (ENUM_TIMEFRAMES)_Period;
|
|
string name = EnumToString(p); // e.g. "PERIOD_M15"
|
|
StringReplace(name, "PERIOD_", ""); // "M15"
|
|
return name;
|
|
}
|
|
|
|
#endif // DATAHARVESTER_MQH
|