# 블로그 기사용 시험 매수 실측 증거 — 격리 임시 PostgreSQL + 합성 호가만 쓴다. # 베이스: reports/trial-buy-independent-review/a26-a35/browser/run_fixture.py # 운영 DB·운영 .env·실계좌·외부 거래소·배포를 쓰지 않는다. 월 한도 환경변수는 **주지 않는다** (제품 기본값 노출). from pathlib import Path import asyncio import json import os import signal import subprocess import sys import tempfile ROOT = Path(__file__).resolve().parents[3] OUT = Path(__file__).resolve().parent PG = Path('/opt/homebrew/opt/postgresql@16/bin') SOCK_PREFIX = 'weltiq-blog-trial-' PG_PORT = '55531' ROLE = DB_NAME = 'weltiq_blog_trial' API_PORT = 8891 BASE = f'http://127.0.0.1:{API_PORT}' FIXTURE_EMAIL = 'blog-trial@example.invalid' FIXTURE_PW = 'Local-Test-Only-9027!' FEED_SOURCE = 'blog-evidence-fixture' META = {'external_run_id': 'blog-a2-02-run-1', 'tool': '29B-review.py'} TAGS = ['a2-02', 'evidence'] TRACK = 'blog-a2-02' ACCOUNT_TABLES = ('account.accounts', 'account.portfolios', 'account.sleeves', 'account.ledger', 'account.direction_leases', 'account.reduce_reservations', 'trading.intents', 'trading.fills', 'trading.order_transitions') def child(): from decimal import Decimal from psycopg import sql import httpx import uvicorn from weltiq.api import auth from weltiq.api.server import ApiState, build_app from weltiq.core import db from weltiq.core.events import BBO, utcnow from weltiq.gateway.gateway import Gateway from weltiq.policy.guard import GuardInputs from weltiq.runner.forward import ForwardEngine from weltiq.simbroker.broker import SimBroker from weltiq.simbroker.profile import PROFILES, BINANCE_SPOT_V1, US_EQUITY_V4 PROFILE = {'binance-spot': BINANCE_SPOT_V1, 'binance-usdm': PROFILES[('binance-usdm', 1)], 'us-equity': US_EQUITY_V4} # 합성 호가 — 런타임에 바꿀 수 있게 dict 하나에 둔다 (T3 의 목표가 발동을 위해 한 번 내린다). QUOTES = {('binance-spot', 'BTCUSDT'): [Decimal('50000'), Decimal('50001')], ('binance-usdm', 'BTCUSDT'): [Decimal('50000'), Decimal('50001')], ('us-equity', 'AAPL'): [Decimal('230.00'), Decimal('230.02')]} ev: dict = {'environment': {}, 'setup': {}, 'cases': [], 'events': []} def log_event(what, **kw): ev['events'].append({'at': utcnow().isoformat(), 'what': what, **kw}) conn = db.connect() auth.ensure(conn) uid = auth.signup(conn, FIXTURE_EMAIL, FIXTURE_PW) auth.mark_verified(conn, uid) conn.execute("UPDATE auth.users SET plan='pro' WHERE user_id=%s", (uid,)) # 상장 행 — 주문 접수·시세 구독 판정이 market.listings 를 읽는다. 무엇을 넣었는지 그대로 기록한다. listings = [('BTCUSDT', 'BTCUSDT', 'crypto', 'Bitcoin / USDT', 'USDT', 'binance-spot', ['binance-usdm'], True, 1), ('AAPL', 'AAPL', 'us_stock', 'Apple Inc.', 'USD', 'us-equity', [], True, 1)] for row in listings: conn.execute("INSERT INTO market.listings (instrument_id,symbol,asset_class,name,quote_ccy," "price_venue,alt_venues,tradable,rank) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)" " ON CONFLICT (instrument_id) DO NOTHING", row) # 계좌 원본 — 비교가 의미를 갖도록 현금 한 줄을 넣는다 (픽스처가 넣은 단 한 줄이다). conn.execute("INSERT INTO account.ledger (txn_id,account_id,book,amount,ccy,ref_kind)" " VALUES ('blog-a2-02-seed',%s,'cash','100000','USDT','blog_evidence_fixture')", (f'acc_{uid}',)) conn.commit() api_key = auth.issue_key(conn, uid, 'blog-a2-02', ['read-data', 'run-forward-tests']) H = {'Authorization': 'Bearer ' + api_key['secret'], 'Content-Type': 'application/json'} ev['setup'] = {'user_id': uid, 'email': FIXTURE_EMAIL, 'plan': 'pro', 'api_key_id': api_key['key_id'], 'api_key_scopes': api_key['scopes'], 'listings_inserted': [{'instrument_id': r[0], 'asset_class': r[2], 'price_venue': r[5], 'alt_venues': r[6], 'tradable': r[7]} for r in listings], 'account_seed': {'account_id': f'acc_{uid}', 'book': 'cash', 'amount': '100000', 'ccy': 'USDT'}, 'synthetic_quotes': {f'{v}:{s}': [str(q[0]), str(q[1])] for (v, s), q in QUOTES.items()}, 'profiles': {v: [p.profile_id, p.version] for v, p in PROFILE.items()}, 'env_forward_test_monthly_limit': os.environ.get('WELTIQ_FORWARD_TEST_MONTHLY_LIMIT', '(unset)')} market_conn = db.connect() engine_conn = db.connect() engine = ForwardEngine(engine_conn, PROFILE) state = ApiState(gateway=Gateway(conn, {v: SimBroker(p) for v, p in PROFILE.items()}), conn=conn, tokens={}, guard_inputs=lambda symbol=None: GuardInputs( now=utcnow(), market_as_of=utcnow(), data_fresh=True, position_qty=Decimal(0), daily_pnl=Decimal(0))) app = build_app(state) snap_conn = db.connect() def usage_rows(): """한도 원장 — 증가와 **보상(delta=-1)** 을 둘 다 그대로 본다.""" out = {} for table, order in (('forward.usage', 'user_id,ym'), ('forward.usage_events', 'id')): q = sql.SQL('SELECT row_to_json(t)::text AS row FROM {} t ORDER BY {}').format( sql.Identifier(*table.split('.')), sql.SQL(order)) out[table] = [json.loads(r['row']) for r in snap_conn.execute(q).fetchall()] snap_conn.commit() return out def account_snapshot(): out = {} for table in ACCOUNT_TABLES: q = sql.SQL('SELECT row_to_json(t)::text AS row FROM {} t ORDER BY row_to_json(t)::text').format( sql.Identifier(*table.split('.'))) out[table] = [json.loads(r['row']) for r in snap_conn.execute(q).fetchall()] snap_conn.commit() return out async def quotes(): while True: at = utcnow() engine.tick() for (venue, symbol), (bid, ask) in QUOTES.items(): market_conn.execute( "INSERT INTO market.quote_latest(venue,symbol,event_time,bid,bid_qty,ask,ask_qty)" " VALUES (%s,%s,%s,%s,10,%s,10) ON CONFLICT(venue,symbol)" " DO UPDATE SET event_time=EXCLUDED.event_time,bid=EXCLUDED.bid,ask=EXCLUDED.ask," "updated_at=now()", (venue, symbol, at, bid, ask)) market_conn.commit() engine.on_event(BBO(venue=venue, symbol=symbol, event_time=at, bid=bid, ask=ask, bid_qty=Decimal(10), ask_qty=Decimal(10), feed_source=FEED_SOURCE)) engine.tick() await asyncio.sleep(1) # ── 케이스 본문 ──────────────────────────────────── def body(cid, venue, symbol, side, exit_plan, memo, **extra): b = {'venue': venue, 'symbol': symbol, 'client_order_id': cid, 'side': side, 'exit_plan': exit_plan, 'max_duration_hours': 24, 'memo': memo, 'client_tracking_id': TRACK, 'tags': TAGS, 'metadata': META} b.update(extra) return b T1_BODY = body('blog-a2-02-t1', 'binance-spot', 'BTCUSDT', 'long', {'close_after_seconds': 60, 'take_profit': '51000', 'stop_loss': '49000'}, '기사용 실측: 현물 롱 60초 자동 완결') async def case(client, cid, what, method, path, *, json_body=None, expect=None, raw=False): r = await client.request(method, path, json=json_body, headers=H) try: payload = r.json() except Exception: payload = None rec = {'id': cid, 'what': what, 'request': {'method': method, 'path': path, 'body': json_body}, 'response': {'status': r.status_code, 'json': payload}, 'at': utcnow().isoformat()} if raw or payload is None: rec['response']['text'] = r.text if expect is not None: rec['expected_status'] = expect rec['status_matches_expectation'] = (r.status_code == expect) ev['cases'].append(rec) return rec async def poll(client, test_id, deadline_s, stop_when): """상세를 2초마다 읽고 상태가 바뀔 때마다 표본을 남긴다.""" import time t0, samples, last = time.monotonic(), [], None while time.monotonic() - t0 < deadline_s: r = await client.get(f'/v1/forward-tests/{test_id}', headers=H) d = r.json() row = (d.get('comparison') or [{}])[0] ac = row.get('auto_close') or {} fp = {'status': row.get('status'), 'auto_close_state': ac.get('state'), 'exit_trigger': row.get('exit_trigger'), 'close_reason': row.get('close_reason'), 'effective_close_basis': ac.get('effective_close_basis'), 'exit_block': ac.get('exit_block'), 'waiting_on': ac.get('waiting_on'), 'blocked': ac.get('blocked'), 'test_status': d.get('status'), 'end_reason': d.get('end_reason')} if fp != last: samples.append({'at': utcnow().isoformat(), 'elapsed_s': round(time.monotonic() - t0, 1), **fp, 'exit_trigger_at': row.get('exit_trigger_at'), 'exit_trigger_px': row.get('exit_trigger_px'), 'entry': row.get('entry'), 'exit': row.get('exit')}) last = fp if stop_when(d, row): return samples, d await asyncio.sleep(2) return samples, d async def cases(): async with httpx.AsyncClient(base_url=BASE, timeout=30) as c: # T0 — 시작 직전 한도 기본값 + 계좌 원본 t0 = await case(c, 'T0', '시작 직전 usage + 계좌 원장 스냅샷', 'GET', '/v1/forward-tests/usage', expect=200) t0['account_snapshot'] = account_snapshot() t0['usage_ledger'] = usage_rows() # T1 — 현물 롱, 60초 + 목표가·손절가 t1 = await case(c, 'T1', 'binance-spot BTCUSDT long · close_after_seconds=60 + TP/SL', 'POST', '/v1/trial-buys', json_body=T1_BODY, expect=201) t1_id = (t1['response']['json'] or {}).get('test_id') # T2 — 같은 호출을 현물 숏으로 await case(c, 'T2', 'binance-spot + side=short → unsupported_direction', 'POST', '/v1/trial-buys', json_body=body('blog-a2-02-t2', 'binance-spot', 'BTCUSDT', 'short', {'close_after_seconds': 60, 'take_profit': '49000', 'stop_loss': '51000'}, '기사용 실측: 현물 숏 거절 원문'), expect=422) # T3 — 무기한 선물 숏 (목표가 < 손절가) t3 = await case(c, 'T3', 'binance-usdm BTCUSDT short · TP 49000 < SL 51000 · close_after 60', 'POST', '/v1/trial-buys', json_body=body('blog-a2-02-t3', 'binance-usdm', 'BTCUSDT', 'short', {'close_after_seconds': 60, 'take_profit': '49000', 'stop_loss': '51000'}, '기사용 실측: 선물 숏 목표가 발동'), expect=201) t3_id = (t3['response']['json'] or {}).get('test_id') # T4 — us-equity AAPL 숏 (기본 라이선스 그대로) await case(c, 'T4', 'us-equity AAPL short — 기본 라이선스(license_pending) 그대로', 'POST', '/v1/trial-buys', json_body=body('blog-a2-02-t4', 'us-equity', 'AAPL', 'short', {'close_after_seconds': 60}, '기사용 실측: 주식 숏 (라이선스 기본값)')) # T4b — 같은 호출을 시장 강제 개방 플래그(내부 검증용)로 다시 os.environ['WELTIQ_MARKET_OPEN'] = 'us-equity' log_event('WELTIQ_MARKET_OPEN=us-equity 설정 (내부 검증용 강제 개방 — 방향 거절과 라이선스 거절을 분리하기 위해)') t4b = await case(c, 'T4b', 'us-equity AAPL short — WELTIQ_MARKET_OPEN=us-equity 로 다시', 'POST', '/v1/trial-buys', json_body=body('blog-a2-02-t4b', 'us-equity', 'AAPL', 'short', {'close_after_seconds': 60}, '기사용 실측: 주식 숏 (시장 개방)')) t4b_id = (t4b['response']['json'] or {}).get('test_id') # T5 — T1 과 같은 client_order_id + 같은 본문 await case(c, 'T5', 'T1 과 같은 client_order_id + 같은 본문 재전송', 'POST', '/v1/trial-buys', json_body=T1_BODY, expect=200) u5 = await case(c, 'T5-usage', '재전송 직후 usage (한도 재사용 확인)', 'GET', '/v1/forward-tests/usage', expect=200) u5['note'] = 'T5 직후 — T0 대비 증가분이 승인된 신규 진입 건수와 같아야 한다' # T6 — 같은 client_order_id + 다른 본문 (memo 한 글자) diff = dict(T1_BODY, memo=T1_BODY['memo'] + '!') await case(c, 'T6', '같은 client_order_id + memo 한 글자만 다른 본문', 'POST', '/v1/trial-buys', json_body=diff, expect=409) # T9 — qty 를 명시 await case(c, 'T9', 'qty 를 명시해서 보내기', 'POST', '/v1/trial-buys', json_body=body('blog-a2-02-t9', 'binance-spot', 'BTCUSDT', 'long', {'close_after_seconds': 60}, '기사용 실측: 수량 지정 거절', qty='0.001'), expect=422) # T10 — 시간 조건 없음 await case(c, 'T10', '시간 조건을 아무것도 안 주기', 'POST', '/v1/trial-buys', json_body=body('blog-a2-02-t10', 'binance-spot', 'BTCUSDT', 'long', {'take_profit': '51000', 'stop_loss': '49000'}, '기사용 실측: 시한 없는 계획 거절'), expect=422) # ── 추적: T3 의 목표가 발동 (진입 체결 뒤 합성 호가를 한 번 내린다) ── if t3_id: entered, _ = await poll(c, t3_id, 60, lambda d, row: (row.get('entry') or {}).get('qty')) ev['cases'].append({'id': 'T3-entry', 'what': 'T3 진입 체결까지', 'samples': entered}) QUOTES[('binance-usdm', 'BTCUSDT')] = [Decimal('48900'), Decimal('48901')] log_event('합성 호가 변경: binance-usdm BTCUSDT 50000/50001 → 48900/48901 ' '(숏 목표가 49000 을 ask 가 통과하게 — 이 한 번의 변경 외에 호가는 고정이다)') s3, d3 = await poll(c, t3_id, 120, lambda d, row: row.get('close_reason')) ev['cases'].append({'id': 'T3-exit', 'what': 'T3 청산 발동까지', 'samples': s3, 'final_detail': d3}) if t1_id: s1, d1 = await poll(c, t1_id, 150, lambda d, row: row.get('close_reason')) ev['cases'].append({'id': 'T1-exit', 'what': 'T1 시간 발동까지', 'samples': s1, 'final_detail': d1}) if t4b_id: s4, d4 = await poll(c, t4b_id, 30, lambda d, row: row.get('status') == 'closed') ev['cases'].append({'id': 'T4b-track', 'what': 'T4b 집행 결과 추적', 'samples': s4, 'final_detail': d4}) # T7 — 상세 + CSV + JSON 내보내기 if t1_id: await case(c, 'T7-detail', 'T1 상세', 'GET', f'/v1/forward-tests/{t1_id}', expect=200) csv_rec = await case(c, 'T7-csv', 'T1 CSV 내보내기', 'GET', f'/v1/forward-tests/{t1_id}/export?format=csv', expect=200, raw=True) text = csv_rec['response'].get('text') or '' csv_rec['csv_header'] = text.splitlines()[0] if text else None csv_rec['csv_lines'] = text.splitlines() await case(c, 'T7-json', 'T1 JSON 내보내기', 'GET', f'/v1/forward-tests/{t1_id}/export?format=json', expect=200) # T8 — 마지막 usage + 계좌 원장 t8 = await case(c, 'T8', '전부 끝난 뒤 usage + 계좌 원장 스냅샷', 'GET', '/v1/forward-tests/usage', expect=200) after = account_snapshot() t8['account_snapshot'] = after t8['account_unchanged_vs_T0'] = (after == t0['account_snapshot']) t8['usage_ledger'] = usage_rows() async def serve(): feed = asyncio.create_task(quotes()) server = uvicorn.Server(uvicorn.Config(app, host='127.0.0.1', port=API_PORT, log_level='warning')) runner = asyncio.create_task(server.serve()) try: for _ in range(100): # 기동 대기 if server.started: break await asyncio.sleep(0.1) ev['environment'] = {'base_url': BASE, 'db': DB_NAME, 'pg_port': PG_PORT, 'commit_sha': os.environ.get('BLOG_EVIDENCE_SHA', '(unknown)'), 'started_at': utcnow().isoformat(), 'note': '격리 임시 PostgreSQL · 합성 호가 · 운영 배포 아님 · 실계좌 아님'} try: await cases() except Exception as exc: # 실패도 증거다 — 숨기지 않는다 import traceback ev['fatal'] = {'error': repr(exc), 'traceback': traceback.format_exc()} ev['environment']['finished_at'] = utcnow().isoformat() (OUT / 'evidence.json').write_text(json.dumps(ev, ensure_ascii=False, indent=1, default=str)) print('evidence.json written', flush=True) os.kill(os.getpid(), signal.SIGINT) # uvicorn 의 SIGINT 핸들러 → graceful shutdown await runner finally: feed.cancel() for c_ in (conn, market_conn, engine_conn, snap_conn): c_.close() print(f'blog trial evidence server {BASE} — isolated DB, synthetic quotes only', flush=True) try: asyncio.run(serve()) except KeyboardInterrupt: # 위에서 우리가 보낸 SIGINT — graceful shutdown 은 이미 끝났다 pass def main(): if '--child' in sys.argv: return child() with tempfile.TemporaryDirectory(prefix=SOCK_PREFIX, dir='/tmp') as tmp: env = {k: os.environ[k] for k in ('PATH', 'HOME', 'TMPDIR', 'LANG') if k in os.environ} dsn = f'host={tmp} port={PG_PORT} user={ROLE} dbname={DB_NAME}' sha = subprocess.run(['git', 'rev-parse', 'HEAD'], cwd=ROOT, text=True, capture_output=True).stdout.strip() env.update(PGHOST=tmp, PGPORT=PG_PORT, PGUSER=ROLE, WELTIQ_DB=DB_NAME, WELTIQ_PG_DSN=dsn, WELTIQ_CELL_DSN=dsn, WELTIQ_TEST_CELL_DSN=dsn, WELTIQ_MAIL_WORKER='0', WELTIQ_HTTP_METER='0', # 월 한도 환경변수는 주지 않는다 BLOG_EVIDENCE_SHA=sha, PYTHONPATH=str(ROOT), PATH=str(PG) + os.pathsep + env.get('PATH', ''), LANG='en_US.UTF-8') data = str(Path(tmp) / 'data') def run(args, name): result = subprocess.run(args, cwd=ROOT, env=env, text=True, capture_output=True) (OUT / name).write_text(result.stdout + result.stderr) result.check_returncode() run([str(PG / 'initdb'), '-D', data, '-U', ROLE, '-A', 'trust'], 'initdb.txt') run([str(PG / 'pg_ctl'), '-D', data, '-l', str(Path(tmp) / 'pg.log'), '-o', f"-k {tmp} -p {PG_PORT} -c listen_addresses=''", '-w', 'start'], 'start.txt') try: run([str(PG / 'psql'), '-d', 'postgres', '-c', 'CREATE ROLE weltiq'], 'role.txt') run([str(PG / 'createdb'), DB_NAME], 'db.txt') run([str(PG / 'psql'), '-v', 'ON_ERROR_STOP=1', '-d', DB_NAME, '-f', 'db/schema.sql'], 'schema.txt') run([str(ROOT / '.venv/bin/python'), 'tools/migrate.py'], 'migrate.txt') return subprocess.call([str(ROOT / '.venv/bin/python'), str(Path(__file__).resolve()), '--child'], cwd=ROOT, env=env) finally: run([str(PG / 'pg_ctl'), '-D', data, '-m', 'fast', '-w', 'stop'], 'stop.txt') if __name__ == '__main__': raise SystemExit(main())