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
|