"""Copy this single file into PythonGO. Market data only; no order functions. Uses the documented BaseStrategy/KLineGenerator API and Python standard library. Default target is the website on the same computer: http://127.0.0.1:18080. """ import json import math import queue import re import sqlite3 import threading import time import uuid from datetime import datetime, timedelta, timezone from pathlib import Path from urllib.error import HTTPError from urllib.request import Request, urlopen from pythongo.base import BaseParams, BaseState, Field from pythongo.classdef import TickData, KLineData from pythongo.core import MarketCenter from pythongo.ui import BaseStrategy from pythongo.utils import KLineGenerator, Scheduler CHINA = timezone(timedelta(hours=8)) ALIASES = 'PG-AGMAIN;PG-PSMAIN;PG-PS2611;PG-PS2704;PG-PS2705;PG-AUMAIN;PG-AG2701;PG-AG2702;PG-AG2703' class Params(BaseParams): exchange: str = Field(default='SHFE;GFEX;GFEX;GFEX;GFEX;SHFE;SHFE;SHFE;SHFE', title='交易所（分号分隔）') instrument_id: str = Field(default='agMain;psMain;ps2611;ps2704;ps2705;auMain;ag2701;ag2702;ag2703', title='合约（顺序须对应）') symbols: str = Field(default=ALIASES, title='网站标的（顺序须对应）') server_url: str = Field(default='http://127.0.0.1:18080', title='网站服务地址') ingest_token: str = Field(default='', title='远程推送令牌（本机可留空）') batch_seconds: float = Field(default=0.25, title='实时转发间隔（秒）', ge=0.05, le=5) history_enabled: bool = Field(default=True, title='启动时补传 M1/D1 历史') bar_time_mode: str = Field(default='end', title='K线时间标签 end 或 start') allow_contract_change: bool = Field(default=False, title='显式允许变更绑定（不拼接旧历史）') spool_directory: str = Field(default='', title='断线队列目录（留空自动创建）') dominant_refresh_seconds: int = Field(default=3600, title='主力映射复查间隔（秒）', ge=1800) class State(BaseState): sent: int = Field(default=0, title='已确认转发条数') pending: int = Field(default=0, title='本机待发条数') retries: int = Field(default=0, title='连续失败次数') last_error: str = Field(default='', title='最近转发错误') main_bindings: str = Field(default='', title='MAIN 实际主力合约') mapping_error: str = Field(default='', title='主力查询错误') def as_datetime(value): dt = value if isinstance(value, datetime) else datetime.fromisoformat(str(value).replace('Z', '+00:00')) return dt.replace(tzinfo=CHINA) if dt.tzinfo is None else dt.astimezone(CHINA) def finite(value, positive=False): try: value = float(value) if not math.isfinite(value) or abs(value) >= 1e100 or (positive and value <= 0): return None return value except (ValueError, TypeError, OverflowError): return None class DurableSender: """Callbacks only enqueue. One worker owns disk and HTTP, even during backoff.""" def __init__(self, url, token, path, interval=.25): self.url = url.rstrip('/') if not self.url.endswith('/api/pythongo/ingest'): self.url += '/api/pythongo/ingest' self.token, self.path, self.interval = token, Path(path), interval self.path.parent.mkdir(parents=True, exist_ok=True) self.input = queue.Queue() self.latest = {} self.heartbeat_sources = {} self.latest_lock = threading.Lock() self.stop_event = threading.Event() self.sent = self.pending = self.retries = 0 self.last_error = '' self.thread = threading.Thread(target=self.run, name='PythonGO-market-forwarder', daemon=True) def start(self): self.thread.start() def enqueue(self, event, coalesce=False): event = dict(event) with self.latest_lock: if event['kind'] == 'register': # Finish the old binding's coalesced updates before its successor. for key in [k for k, value in self.latest.items() if value['symbol'] == event['symbol']]: self.input.put(self.latest.pop(key)) event.setdefault('binding_id', uuid.uuid4().hex) self.heartbeat_sources[event['symbol']] = { k:event[k] for k in ('symbol','exchange','instrument_id','binding_id')} else: binding = self.heartbeat_sources.get(event['symbol']) if binding: if (event['exchange'].upper(),event['instrument_id'].lower()) != (binding['exchange'].upper(),binding['instrument_id'].lower()): return # Ignore callbacks from a retired binding. event['binding_id'] = binding['binding_id'] key = self.key(event) if coalesce: self.latest[key] = event else: self.input.put(event) @staticmethod def key(e): prefix = '|'.join((e['symbol'],e['exchange'].upper(),e['instrument_id'].lower(),e.get('binding_id',''))) if e['kind'] == 'bar': return '|'.join((prefix, 'bar', e['interval'], str(e['bar']['time']))) if e['kind'] == 'history_complete': return prefix+'|history|'+e['interval'] return prefix+'|'+e['kind'] def post(self, body): headers = {'Content-Type': 'application/json'} if self.token: headers['Authorization'] = 'Bearer '+self.token request = Request(self.url, data=json.dumps(body, ensure_ascii=False, allow_nan=False).encode(), headers=headers) with urlopen(request, timeout=5) as response: result = json.load(response) if not result.get('ok'): raise RuntimeError('接收端没有确认这批行情') def run(self): try: self._run() except Exception as error: self.last_error = '转发线程停止：'+type(error).__name__+'；请检查队列目录/磁盘后重启策略' def _run(self): db = sqlite3.connect(self.path, timeout=15) db.executescript('PRAGMA journal_mode=WAL; CREATE TABLE IF NOT EXISTS outbox(id INTEGER PRIMARY KEY, k TEXT UNIQUE, payload TEXT);') next_send, flight, heartbeat_at = 0, None, 0 try: while True: incoming = [] while len(incoming) < 4000: try: incoming.append(self.input.get_nowait()) except queue.Empty: break with self.latest_lock: # A new binding's tick must not overtake a queued register. if self.input.empty(): incoming.extend(self.latest.values());self.latest.clear() if time.monotonic()-heartbeat_at > 5: incoming.extend(dict(kind='heartbeat',**v) for v in self.heartbeat_sources.values()) heartbeat_at = time.monotonic() for e in incoming: key, text = self.key(e), json.dumps(e, ensure_ascii=False, allow_nan=False) old = db.execute('SELECT payload FROM outbox WHERE k=?', (key,)).fetchone() if old and e['kind'] == 'bar': prior = json.loads(old[0]) if prior['bar']['confirmed'] and not e['bar']['confirmed']: continue db.execute('INSERT INTO outbox(k,payload) VALUES(?,?) ON CONFLICT(k) DO UPDATE SET payload=excluded.payload', (key,text)) db.commit() self.pending = db.execute('SELECT count(*) FROM outbox').fetchone()[0] if self.stop_event.is_set(): if not self.input.empty() or self.latest: continue break # Everything remaining is durable; resume on the next strategy start. if time.monotonic() >= next_send: if flight is None: rows = db.execute('SELECT id,payload FROM outbox ORDER BY id LIMIT 300').fetchall() if rows: flight = (rows, {'batch_id': uuid.uuid4().hex, 'events': [json.loads(r[1]) for r in rows]}) if flight: try: self.post(flight[1]) # Newer updates sharing the same row must survive an in-flight acknowledgement. for row in flight[0]: db.execute('DELETE FROM outbox WHERE id=? AND payload=?', row) db.commit();self.sent += len(flight[0]);flight = None self.retries = 0;self.last_error = '';next_send = time.monotonic() except Exception as error: self.retries += 1 code = error.code if isinstance(error, HTTPError) else None self.last_error = ('HTTP '+str(code) if code else type(error).__name__)+': 转发失败，行情保留在磁盘队列' next_send = time.monotonic()+min(2**min(self.retries-1,5),30) self.stop_event.wait(self.interval) except Exception as error: self.last_error = '转发线程停止：'+type(error).__name__+'；请检查队列目录/磁盘后重启策略' finally: db.commit();db.close() def stop(self): self.stop_event.set();self.thread.join(timeout=8) class MarketForwarder(BaseStrategy): """Only subscribes and forwards Tick/M1/D1 data. Never places or cancels orders.""" def __init__(self): super().__init__() self.params_map, self.state_map = Params(), State() self.generators, self.bindings, self.history_loading = {}, {}, set() self.latest_trading_day = {} self.sender = None self._active = False self._last_heartbeat = self._last_status = 0 self._last_logged_error = '' self._binding_lock = threading.RLock() self._refresh_lock = threading.Lock() self._subscriptions = set() self._generator_tokens = {} self._configured = [] self._resolved = {} self._seen_trading_days = {} self._next_resolution = 0 self._resolution_due = False self._scheduler = None self._market_center = None def on_start(self): p = self.params_map exchanges, instruments, symbols = ([x.strip() for x in value.split(';')] for value in (p.exchange,p.instrument_id,p.symbols)) if not (len(exchanges) == len(instruments) == len(symbols)) or not all(exchanges+instruments+symbols): raise ValueError('交易所、合约和网站标的数量必须一致且不能为空') if len(set(symbols)) != len(symbols): raise ValueError('网站标的不能重复') if p.bar_time_mode not in ('start','end'): raise ValueError('bar_time_mode 只能是 start 或 end') if self.sender and self.sender.thread.is_alive(): raise RuntimeError('上次转发线程尚未停止，请稍后再启动') directory = Path(p.spool_directory) if p.spool_directory else Path(__file__).resolve().parent/'.market-forwarder' path = directory/('outbox-'+str(self.strategy_id)+'.sqlite3') self.sender = DurableSender(p.server_url,p.ingest_token,path,p.batch_seconds) self._active = True;self.generators = {};self.bindings = {};self.history_loading = set() self._resolved = {};self._subscriptions = set();self._generator_tokens = {} self._seen_trading_days = {};self._resolution_due = False self._configured = [(symbol,exchange.upper(),instrument) for exchange,instrument,symbol in zip(exchanges,instruments,symbols)] self._market_center = MarketCenter() self.sender.start() self.refresh_dominants(initial=True) # The override subscribes resolved contracts, keeping MAIN in the saved parameters. super().on_start() self.trading = False self._scheduler = Scheduler() # Private scheduler; never stop PythonGO's shared scheduler. self._scheduler.add_job(self.maintenance, trigger='interval', seconds=5, id='market-forwarder') self._scheduler.start() self.output('行情转发已启动；接收地址：',p.server_url,'；本策略不发送交易指令。') def refresh_dominants(self, initial=False): if not self._active or not self._refresh_lock.acquire(blocking=False): return try: lookup, errors = {}, [] for exchange in sorted({ex for _,ex,ins in self._configured if ins.lower().endswith('main')}): try: values = self._market_center.get_dominant_list(exchange) if not isinstance(values, list): raise ValueError('主力接口未返回合约列表') lookup[exchange] = [str(v).strip() for v in values] except Exception as error: errors.append(exchange+' 主力查询失败：'+str(error)) desired = dict(self._resolved) for symbol,exchange,logical in self._configured: instrument = logical if logical.lower().endswith('main'): product = logical[:-4] matches = [ins for ins in lookup.get(exchange,[]) if re.fullmatch(re.escape(product)+r'\d{3,4}',ins,re.I)] if len(matches) != 1: errors.append(logical+' 未获得唯一实际主力合约；保留已有绑定，未首次解析的标的等待重试') continue instrument = matches[0] desired[symbol] = (exchange,instrument) with self._binding_lock: if not self._active: return changed = [symbol for symbol,value in desired.items() if tuple(part.lower() for part in value) != tuple(part.lower() for part in self._resolved.get(symbol,()))] prior_keys = set(self.bindings) self.bindings = {} for symbol,(exchange,instrument) in desired.items(): self.bindings.setdefault((exchange,instrument.lower()),[]).append((symbol,exchange,instrument)) logical_by_symbol = {symbol:logical for symbol,_,logical in self._configured} for symbol in changed: exchange,instrument = desired[symbol] logical = logical_by_symbol[symbol] main = logical.lower().endswith('main') self.sender.enqueue(dict(kind='register',symbol=symbol,exchange=exchange,instrument_id=instrument, logical_instrument_id=logical, mapping_source='PythonGO.MarketCenter.get_dominant_list' if main else 'configured', allow_contract_change=main or self.params_map.allow_contract_change)) if main: old = self._resolved.get(symbol,('', '未绑定'))[1] self.output('主力映射：',symbol,old,'→',instrument) self._resolved = desired reload_keys = {(desired[s][0],desired[s][1].lower()) for s in changed} for key in reload_keys: self.load_history(key) if not initial: for key in set(self.bindings)-prior_keys: self.sub_market_data(*key) for key in prior_keys-set(self.bindings): self.unsub_market_data(*key) self.generators.pop(key,None);self._generator_tokens.pop(key,None) self.state_map.main_bindings = '; '.join(symbol+'='+desired[symbol][1] for symbol,_,logical in self._configured if logical.lower().endswith('main') and symbol in desired) message = '; '.join(errors) if message and message != self.state_map.mapping_error: self.output(message) self.state_map.mapping_error = message self._next_resolution = time.monotonic()+(300 if errors else self.params_map.dominant_refresh_seconds) self._resolution_due = False finally: self._refresh_lock.release() def load_history(self, key): exchange,instrument = self.bindings[key][0][1:] if key not in self.generators: self.generators[key] = [] token = self._generator_tokens[key] = uuid.uuid4().hex for interval,style in [('1m','M1'),('1d','D1')]: try: generator = KLineGenerator(exchange=exchange,instrument_id=instrument,style=style, callback=lambda k,key=key,interval=interval,token=token:self.bar(key,interval,k,True,token), real_time_callback=lambda k,key=key,interval=interval,token=token:self.bar(key,interval,k,False,token)) self.generators[key].append((interval,generator)) except Exception as error: self.output(instrument,style,'K线合成器初始化失败：',str(error)) by_interval = dict(self.generators[key]) for interval in ('1m','1d'): available = False self.history_loading.add((key,interval)) try: if self.params_map.history_enabled and interval in by_interval: by_interval[interval].push_history_data();available = True except Exception as error: self.output(instrument,interval,'历史加载失败，保留实时订阅：',str(error)) finally: self.history_loading.discard((key,interval)) for symbol,ex,ins in self.bindings[key]: self.sender.enqueue(dict(kind='history_complete',symbol=symbol,exchange=ex,instrument_id=ins, interval=interval,available=available)) def sub_market_data(self, exchange=None, instrument_id=None): if exchange is None or instrument_id is None: for ex,ins in self.bindings: self.sub_market_data(ex,ins) else: key = (exchange.upper(),instrument_id.lower()) if key not in self._subscriptions: super().sub_market_data(exchange=exchange,instrument_id=instrument_id) self._subscriptions.add(key) def unsub_market_data(self, exchange=None, instrument_id=None): if exchange is None or instrument_id is None: for ex,ins in list(self._subscriptions): self.unsub_market_data(ex,ins) else: key = (exchange.upper(),instrument_id.lower()) if key in self._subscriptions: super().unsub_market_data(exchange=exchange,instrument_id=instrument_id) self._subscriptions.discard(key) def maintenance(self): if not self._active: return if self._resolution_due or time.monotonic() >= self._next_resolution: self.refresh_dominants() for attr in ('sent','pending','retries','last_error'): setattr(self.state_map,attr,getattr(self.sender,attr)) if self.sender.last_error and self.sender.last_error != self._last_logged_error: self.output(self.sender.last_error);self._last_logged_error = self.sender.last_error self.update_status_bar() def bar(self, key, interval, kline: KLineData, completed, token=None): with self._binding_lock: if token is not None and self._generator_tokens.get(key) != token: return self._bar(key,interval,kline,completed) def _bar(self, key, interval, kline, completed): if not self._active or not self.sender: return dt = as_datetime(kline.datetime) historical = (key,interval) in self.history_loading if interval == '1m': stamp = int(dt.timestamp())-(60 if self.params_map.bar_time_mode == 'end' else 0) stamp = stamp//60*60 confirmed = bool(completed and stamp+60 <= time.time()) date = datetime.fromtimestamp(stamp,CHINA).strftime('%Y-%m-%d') else: if not completed and self.latest_trading_day.get(key): dt = datetime.strptime(self.latest_trading_day[key],'%Y%m%d').replace(tzinfo=CHINA) day = dt.replace(hour=0,minute=0,second=0,microsecond=0) stamp, date = int(day.timestamp()), day.strftime('%Y-%m-%d') confirmed = bool(completed and stamp+15*3600 <= time.time()) values = {k:finite(getattr(kline,k,None),k!='volume') for k in ('open','high','low','close','volume')} if any(v is None for v in values.values()) or values['volume'] < 0: return bar = dict(time=stamp,date=date,confirmed=confirmed,**values) if interval == '1d': bar['end_time'] = stamp+15*3600 for symbol,exchange,instrument in self.bindings.get(key,[]): self.sender.enqueue(dict(kind='bar',symbol=symbol,exchange=exchange,instrument_id=instrument, interval=interval,bar=bar,historical=historical,observed_at=time.time()),coalesce=not completed) def on_tick(self, tick: TickData): super().on_tick(tick) with self._binding_lock: self._tick(tick) def _tick(self, tick): if not self._active: return key = (str(tick.exchange).upper(),str(tick.instrument_id).lower()) if key not in self.bindings: return day = str(getattr(tick,'trading_day','')).replace('-','') if len(day)==8 and day.isdigit(): self.latest_trading_day[key] = day prior_day = self._seen_trading_days.get(key[0]) if prior_day and day > prior_day: self._resolution_due = True if not prior_day or day > prior_day: self._seen_trading_days[key[0]] = day for _,generator in self.generators.get(key,[]): generator.tick_to_kline(tick) price = finite(tick.last_price,True) if price is None: return quote = dict(price=price,time=as_datetime(tick.datetime).timestamp(),trading_day=day) for target,source in [('bid','bid_price1'),('ask','ask_price1'),('previous','pre_close_price'), ('open','open_price'),('high','high_price'),('low','low_price'),('volume','volume')]: val = finite(getattr(tick,source,None),target!='volume') if val is not None: quote[target] = val now = time.monotonic() for symbol,exchange,instrument in self.bindings[key]: self.sender.enqueue(dict(kind='tick',symbol=symbol,exchange=exchange,instrument_id=instrument,tick=quote),coalesce=True) if now-self._last_heartbeat > 5: for bindings in self.bindings.values(): for symbol,exchange,instrument in bindings: self.sender.enqueue(dict(kind='heartbeat',symbol=symbol,exchange=exchange,instrument_id=instrument),coalesce=True) self._last_heartbeat = now if now-self._last_status > 1: for attr in ('sent','pending','retries','last_error'): setattr(self.state_map,attr,getattr(self.sender,attr)) if self.sender.last_error and self.sender.last_error != self._last_logged_error: self.output(self.sender.last_error);self._last_logged_error = self.sender.last_error self.update_status_bar();self._last_status = now def on_stop(self): with self._binding_lock: self._active = False if self._scheduler: self._scheduler.stop();self._scheduler = None # BaseStrategy.on_stop clears the shared PythonGO scheduler, including other strategies. self.trading = False self.save_instance_file() self.unsub_market_data() if self.sender: self.sender.stop() self.update_status_bar() self.output('行情转发已停止')