Spaces:
Running
Running
File size: 5,782 Bytes
9831ced | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | """Poll real closed candles with the same selectable dashboard pipeline."""
import argparse
import csv
from dataclasses import asdict
import json
from pathlib import Path
import time
import pandas as pd
from config import MODEL_NAMES, OUTPUT_DIR, PROVIDERS, TIMEFRAMES
from pipeline import RunConfig, live_result, prepare_history
from utils.helpers import price_direction
def main():
parser = argparse.ArgumentParser(description='MYTHOS selectable closed-candle forecast logger.')
parser.add_argument('--provider', choices=PROVIDERS, default='Yahoo Finance')
parser.add_argument('--market', choices=['Forex', 'Crypto'], default='Crypto')
parser.add_argument('--symbol', default='BTC-USD')
parser.add_argument('--timeframe', choices=TIMEFRAMES, default='5m')
parser.add_argument('--models', nargs='+', choices=MODEL_NAMES, default=['ARIMA-GARCH'])
parser.add_argument('--lookback-days', type=int, default=30)
parser.add_argument('--window', type=int, default=200)
parser.add_argument('--horizon', type=int, default=1)
parser.add_argument('--features', action='store_true')
parser.add_argument('--poll-seconds', type=int, default=60)
args = parser.parse_args()
config = RunConfig(provider=args.provider, market=args.market, symbol=args.symbol,
timeframe=args.timeframe, models=tuple(args.models),
mode='Parallel models' if len(args.models) > 1 else 'Single model',
lookback_days=args.lookback_days, window=args.window,
horizon=args.horizon, use_features=args.features)
run_id = pd.Timestamp.now(tz='UTC').strftime('%Y%m%dT%H%M%SZ')
Path(OUTPUT_DIR).mkdir(parents=True, exist_ok=True)
path = Path(OUTPUT_DIR) / f'live_logger_{run_id}.csv'
path.with_suffix('.json').write_text(json.dumps(asdict(config), indent=2), encoding='utf-8')
fields = ['model', 'predicted_at_utc', 'origin_open_utc', 'target_open_utc', 'current_close',
'predicted_close', 'predicted_direction', 'entry', 'stop_loss', 'take_profit',
'risk_reward', 'actual_close', 'actual_direction', 'correct', 'status']
def log(row):
new_file = not path.exists()
with path.open('a', newline='', encoding='utf-8') as file:
writer = csv.DictWriter(file, fieldnames=fields)
if new_file:
writer.writeheader()
writer.writerow(row)
pending = []
last_origin = None
print(f'MYTHOS {config.provider} · {config.symbol} · {config.timeframe} · {config.models}. Logging: {path}')
try:
while True:
try:
frame, features = prepare_history(config)
still_pending = []
for row in pending:
target = pd.Timestamp(row['target_open_utc'])
if target in frame.index:
actual = float(frame.loc[target, 'Close'])
direction = price_direction(actual, row['current_close'])
interval = frame.loc[pd.Timestamp(row['origin_open_utc']):target].index.to_series().diff().dropna()
continuous = len(interval) == config.horizon and (interval == config.candle_step).all()
verified = {**row, 'actual_close': actual, 'actual_direction': direction,
'correct': direction == row['predicted_direction'] if continuous else '',
'status': 'VERIFIED' if continuous else 'GAP — not scored'}
log(verified)
print(verified)
elif frame.index[-1] > target:
log({**row, 'status': 'MISSING TARGET — not scored'})
else:
still_pending.append(row)
pending = still_pending
if last_origin != frame.index[-1]:
result = live_result(config, frame, features)
last_origin = frame.index[-1]
for forecast in result['forecasts']:
if not forecast['forecast_path'] or result['metadata']['stale']:
print(f"{forecast['Model']}: {forecast['Status']} {forecast['Error']}")
continue
pending.append({
'model': forecast['Model'], 'predicted_at_utc': result['metadata']['fetched_at_utc'],
'origin_open_utc': last_origin.isoformat(),
'target_open_utc': (last_origin + config.candle_step * config.horizon).isoformat(),
'current_close': forecast['Current close'], 'predicted_close': forecast['Predicted close'],
'predicted_direction': forecast['Direction'].lower(),
'entry': result['signal'].get('trade_levels', {}).get('entry'),
'stop_loss': result['signal'].get('trade_levels', {}).get('stop_loss'),
'take_profit': result['signal'].get('trade_levels', {}).get('take_profit'),
'risk_reward': result['signal'].get('trade_levels', {}).get('risk_reward'),
})
print(f"{forecast['Model']}: {forecast['Direction']} → {forecast['Predicted close']:.8f}")
except Exception as error:
print(f'Pipeline error: {error}')
time.sleep(max(1, args.poll_seconds))
except KeyboardInterrupt:
for row in pending:
log({**row, 'status': 'STOPPED BEFORE TARGET — not scored'})
print('Stopped.')
if __name__ == '__main__':
main()
|