hyowons/openclaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

auto: 일일 백업 2026-08-01 02:00

Browse Source 
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
hyowons
2026-08-01 02:00:02 +09:00
parent d7042430c3
commit 30be97ab2f
177 changed files with 4124 additions and 3489 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agents/stock/workspace/scripts/orders/tests/test_trailing.py
+363
View File
@@ -0,0 +1,363 @@
"""계단식 트레일링 단위테스트 (2026-07-31).
세 층을 본다:
1. 순수 함수 — 누적→추가 환산, 수량 배분, 계단별 손절선, 상향 판단
2. 감시 루프 — 레그 체결/소멸 판정과 알림 (키움 호출은 전부 stub, 상태파일은 임시경로로 격리)
3. JS 대조 — 렌더된 페이지의 미리보기 계산이 서버 계산과 같은 값을 내는지 (node 있을 때만)
3번이 특히 중요하다. 미리보기와 발주값이 어긋나면 관리자님이 본 것과 다른 주문이 나간다.
"""
from __future__ import annotations
import json
import random
import shutil
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from orders import trailing as t
class NormalizeStepsTests(unittest.TestCase):
"""입력한 누적 비중 → 내부 추가 비중."""
def test_cumulative_to_incremental(self):
st = t.normalize_steps([{'pct': 10, 'cum': 20}, {'pct': 20, 'cum': 50},
{'pct': 30, 'cum': 100}])
self.assertEqual([s['weight'] for s in st], [20.0, 30.0, 50.0])
self.assertEqual([s['n'] for s in st], [1, 2, 3])
self.assertEqual([s['pct'] for s in st], [10.0, 20.0, 30.0])
def test_single_step(self):
st = t.normalize_steps([{'pct': 8, 'cum': 100}])
self.assertEqual([(s['pct'], s['weight']) for s in st], [(8.0, 100.0)])
def test_partial_liquidation_allowed(self):
"""마지막 누적이 100 미만이어도 통과 — 일부만 계단 청산하고 나머지는 계속 보유."""
st = t.normalize_steps([{'pct': 10, 'cum': 30}, {'pct': 20, 'cum': 60}])
self.assertEqual([s['weight'] for s in st], [30.0, 30.0])
def test_rejects(self):
cases = [
('빈 입력', [], '비어'),
('계단 초과', [{'pct': i, 'cum': i * 10} for i in range(1, 7)], '최대'),
('하락률 역순', [{'pct': 20, 'cum': 30}, {'pct': 10, 'cum': 60}], '깊지 않음'),
('하락률 동일', [{'pct': 10, 'cum': 30}, {'pct': 10, 'cum': 60}], '깊지 않음'),
('누적 역순', [{'pct': 10, 'cum': 60}, {'pct': 20, 'cum': 30}], '크지 않음'),
('누적 100 초과', [{'pct': 10, 'cum': 50}, {'pct': 20, 'cum': 120}], '100'),
('하락률 상한 초과', [{'pct': 40, 'cum': 100}], '범위'),
('하락률 하한 미만', [{'pct': 0.1, 'cum': 100}], '범위'),
]
for label, raw, needle in cases:
with self.subTest(label):
with self.assertRaises(ValueError) as cm:
t.normalize_steps(raw)
self.assertIn(needle, str(cm.exception))
class AllocateStepQtyTests(unittest.TestCase):
"""계단별 주수 배분 — 최대잔여법."""
def test_exact(self):
self.assertEqual(t.allocate_step_qty(100, [20, 30, 50]), [20, 30, 50])
self.assertEqual(t.allocate_step_qty(1943, [20, 30, 50]), [389, 583, 971])
def test_remainder_goes_to_largest_fraction(self):
self.assertEqual(t.allocate_step_qty(7, [20, 30, 50]), [1, 2, 4])
def test_small_qty_keeps_shallow_step_alive(self):
"""2주를 3계단에 나누면 [0,1,1] — '마지막에 몰아주기'였다면 [0,0,2]로 1계단만 남는다."""
self.assertEqual(t.allocate_step_qty(2, [20, 30, 50]), [0, 1, 1])
self.assertEqual(t.allocate_step_qty(1, [20, 30, 50]), [0, 0, 1])
self.assertEqual(t.allocate_step_qty(0, [20, 30, 50]), [0, 0, 0])
def test_sum_preserved_exhaustive(self):
"""전량(합 100%) 조합은 배분 합이 정확히 보유수량이어야 한다."""
for total in range(1, 400):
for w in ([20, 30, 50], [30, 30, 40], [10, 20, 30, 40], [50, 50], [100]):
got = t.allocate_step_qty(total, w)
self.assertEqual(sum(got), total, f'total={total} w={w} → {got}')
self.assertTrue(all(x >= 0 for x in got))
def test_partial_never_exceeds_holding(self):
for total in range(1, 400):
self.assertLessEqual(sum(t.allocate_step_qty(total, [30, 30])), total)
class ComputeLevelsTests(unittest.TestCase):
def test_step_prices(self):
st = t.normalize_steps([{'pct': 10, 'cum': 20}, {'pct': 20, 'cum': 50},
{'pct': 30, 'cum': 100}])
lv = t.compute_step_levels(5000, st)
self.assertEqual([x['cond_uv'] for x in lv], [4500, 4000, 3500])
self.assertEqual([x['ord_uv'] for x in lv], [4490, 3990, 3490]) # 각 2틱 아래
def test_float_noise_removed(self):
"""5500×(1−30/100) 이 3849.9999999999995 라 int() 가 3849 로 깎고
호가내림이 3845 까지 끌어내리던 버그(2026-07-31 수정). 의도는 3850."""
self.assertEqual(t.compute_levels(5500, 30)['cond_uv'], 3850)
self.assertEqual(t.compute_levels(5500, 10)['cond_uv'], 4950)
self.assertEqual(t.compute_levels(4870, 10)['cond_uv'], 4380)
def test_no_tick_loss_when_exact(self):
"""정수로 딱 떨어지는 (고점,폭)은 호가단위 내림 결과와 정확히 같아야 한다."""
for peak in range(1000, 300000, 977):
for pct in (5, 10, 15, 20, 25, 30):
exact = peak * (100 - pct) / 100.0
if abs(exact - round(exact)) > 1e-9:
continue
want = t.floor_to_tick(int(round(exact)))
self.assertEqual(t.compute_levels(peak, pct)['cond_uv'], want,
f'peak={peak} pct={pct}')
def test_floor_ties_all_steps_then_they_spread(self):
"""최저 매도가는 모든 계단에 같은 하한 → 처음엔 뭉치고, 고점이 오르면 다시 벌어진다.
(그래서 주문을 병합하면 안 된다 — 정정만으로는 다시 못 쪼갠다.)"""
st = t.normalize_steps([{'pct': 10, 'cum': 20}, {'pct': 20, 'cum': 50},
{'pct': 30, 'cum': 100}])
tied = t.compute_step_levels(5000, st, min_sell_price=4600)
self.assertEqual([x['cond_uv'] for x in tied], [4600, 4600, 4600])
self.assertTrue(all(x['floor_applied'] for x in tied))
spread = t.compute_step_levels(6000, st, min_sell_price=4600)
self.assertEqual([x['cond_uv'] for x in spread], [5400, 4800, 4600])
class NextStepLevelsTests(unittest.TestCase):
def setUp(self):
st = t.normalize_steps([{'pct': 10, 'cum': 20}, {'pct': 20, 'cum': 50},
{'pct': 30, 'cum': 100}])
self.res = {'peak': 5000, 'min_sell_price': None,
'steps': [dict(s, cond_uv=c, ord_uv=o) for s, c, o in
zip(st, [4500, 4000, 3500], [4490, 3990, 3490])]}
def test_no_new_high(self):
for price in (0, 4900, 5000):
self.assertIsNone(t.next_step_levels(self.res, price))
def test_all_steps_raise(self):
up = t.next_step_levels(self.res, 6000)
self.assertEqual(up['peak'], 6000)
self.assertEqual([(s['n'], s['cond_uv']) for s in up['steps']],
[(1, 5400), (2, 4800), (3, 4200)])
def test_peak_updates_even_when_no_step_moves(self):
"""호가단위 내림 탓에 올릴 조건단가가 없어도 고점은 갱신돼야 한다(정정 콜은 0)."""
res = {'peak': 5000, 'min_sell_price': None,
'steps': [{'n': 1, 'pct': 10.0, 'cum': 100.0, 'weight': 100.0,
'cond_uv': 4500, 'ord_uv': 4490}]}
up = t.next_step_levels(res, 5001)
self.assertEqual(up['peak'], 5001)
self.assertEqual(up['steps'], [])
def test_floor_dominant_then_overtaken(self):
res = {'peak': 5000, 'min_sell_price': 4600,
'steps': [{'n': 1, 'pct': 10.0, 'cum': 100.0, 'weight': 100.0,
'cond_uv': 4600, 'ord_uv': 4590}]}
self.assertEqual(t.next_step_levels(res, 5050)['steps'], []) # 아직 floor 지배
self.assertEqual([(s['n'], s['cond_uv'])
for s in t.next_step_levels(res, 5200)['steps']], [(1, 4680)])
def test_stop_never_moves_down(self):
"""무작위 시세에도 손절선은 단조 증가여야 한다 (상향 전용 불변식)."""
cur = {'peak': 5000, 'min_sell_price': None,
'steps': [{'n': 1, 'pct': 10.0, 'cum': 100.0, 'weight': 100.0,
'cond_uv': 4500, 'ord_uv': 4490}]}
rnd = random.Random(7)
prev = 4500
for _ in range(3000):
nx = t.next_step_levels(cur, rnd.randint(3000, 9000))
if nx is None:
continue
cur['peak'] = nx['peak']
for s in nx['steps']:
self.assertGreaterEqual(s['cond_uv'], prev)
prev = s['cond_uv']
cur['steps'][0]['cond_uv'] = s['cond_uv']
class MonitorFlowTests(unittest.TestCase):
"""감시 루프 — 레그 단위 생존 판정·정정·알림. 키움 호출은 stub."""
def setUp(self):
self.tmp = Path(tempfile.mkdtemp()) / 'trailing_stops.json'
self._orig = (t.STATE_FILE, t._LOCK_FILE)
t.STATE_FILE = self.tmp
t._LOCK_FILE = self.tmp.with_suffix('.json.lock')
import trailing_monitor as tm
self.tm = tm
self.sent = []
self.state = {'open': {}, 'quote': 0, 'execs': [], 'modify': []}
self._orig_fns = (tm.kc.get_open_orders, tm.kc.get_watchlist_quotes,
tm.kc.get_order_executions, tm.kiwoom_order.modify_order,
tm.send_telegram)
tm.kc.get_open_orders = lambda acct, side=None: list(self.state['open'].get(acct, []))
tm.kc.get_watchlist_quotes = lambda codes, exchange='AL': {
c: {'price': self.state['quote']} for c in codes}
tm.kc.get_order_executions = lambda acct: list(self.state['execs'])
tm.kiwoom_order.modify_order = self._fake_modify
tm.send_telegram = lambda msg, parse_mode=None: self.sent.append(msg)
st = t.normalize_steps([{'pct': 10, 'cum': 20}, {'pct': 20, 'cum': 50},
{'pct': 30, 'cum': 100}])
lv = t.compute_step_levels(5500, st, None)
qtys = t.allocate_step_qty(1943, [s['weight'] for s in st])
legs = [dict(s, qty=q, ord_no=f'A{s["n"]:04d}') for s, q in zip(lv, qtys)]
self.res = t.register_steps(account='일반', symbol='381620',
symbol_name='제닉스로보틱스', total_qty=1943,
peak=5500, steps=legs)
self.state['open']['일반'] = [{'ord_no': s['ord_no'], 'unfilled_qty': s['qty'],
'routing_suffix': ''} for s in legs]
def tearDown(self):
(self.tm.kc.get_open_orders, self.tm.kc.get_watchlist_quotes,
self.tm.kc.get_order_executions, self.tm.kiwoom_order.modify_order,
self.tm.send_telegram) = self._orig_fns
t.STATE_FILE, t._LOCK_FILE = self._orig
shutil.rmtree(self.tmp.parent, ignore_errors=True)
def _fake_modify(self, account_label, orig_ord_no, symbol, modify_qty, modify_price,
routing_suffix='', dry_run=False, card_id=None, modify_cond_price=None):
self.state['modify'].append({'orig': orig_ord_no, 'cond': modify_cond_price})
new_no = f'N{len(self.state["modify"]):04d}'
for row in self.state['open'].get(account_label, []):
if row['ord_no'] == orig_ord_no:
row['ord_no'] = new_no # 정정하면 키움이 새 주문번호를 발급한다
return {'ok': True, 'new_ord_no': new_no}
def test_registered_shape(self):
r = t.get(self.res['id'])
self.assertEqual([s['cond_uv'] for s in r['steps']], [4950, 4400, 3850])
self.assertEqual(sum(s['qty'] for s in r['steps']), 1943)
def test_price_drop_does_nothing(self):
self.state['quote'] = 5200
self.tm.check(force=True)
self.assertEqual(self.state['modify'], [])
self.assertEqual(t.get(self.res['id'])['peak'], 5500)
self.assertEqual(self.sent, [])
def test_new_high_raises_every_leg_and_rotates_ord_no(self):
self.state['quote'] = 6000
self.tm.check(force=True)
self.assertEqual([c['cond'] for c in self.state['modify']], [5400, 4800, 4200])
r = t.get(self.res['id'])
self.assertEqual(r['peak'], 6000)
# ord_no 를 갱신하지 않으면 다음 정정이 전부 실패한다 — 최대 함정
self.assertEqual([s['ord_no'] for s in r['steps']], ['N0001', 'N0002', 'N0003'])
self.assertEqual([s['modify_count'] for s in r['steps']], [1, 1, 1])
self.assertEqual(self.sent, []) # 상향은 조용히
def test_one_leg_fills_others_keep_watching(self):
self.state['quote'] = 5500
gone = self.state['open']['일반'].pop(0)
self.state['execs'] = [{'ord_no': gone['ord_no'], 'cntr_qty': 389, 'cntr_uv': 4948}]
self.tm.check(force=True)
r = t.get(self.res['id'])
self.assertEqual([s['n'] for s in r['steps']], [2, 3])
self.assertEqual(len(self.sent), 1)
self.assertIn('체결 389주', self.sent[0])
self.assertIn('남은 계단 2개', self.sent[0])
def test_expiry_not_misread_as_fill(self):
"""1단계가 체결된 날 나머지가 장 마감으로 소멸 — 종목 매도기록이 있어도 소멸로 갈려야 한다.
(종목 단위 ka10170 으로 판정하면 여기서 '체결'로 오판한다.)"""
self.state['quote'] = 5500
self.state['execs'] = [{'ord_no': 'A0001', 'cntr_qty': 389, 'cntr_uv': 4948}]
self.state['open']['일반'] = []
self.tm.check(force=True)
self.assertIsNone(t.get(self.res['id']))
self.assertEqual(len(self.sent), 1) # 레그마다가 아니라 예약 단위 1건
self.assertIn('미체결 소멸', self.sent[0])
self.assertIn('다시 등록', self.sent[0])
def test_open_order_query_failure_preserves_reservation(self):
"""조회 실패를 '사라짐'으로 단정하면 살아있는 예약을 지운다."""
def boom(acct, side=None):
raise RuntimeError('조회 실패')
self.tm.kc.get_open_orders = boom
self.tm.check(force=True)
self.assertEqual(len(t.get(self.res['id'])['steps']), 3)
self.assertEqual(self.sent, [])
def _node_available() -> bool:
return shutil.which('node') is not None
@unittest.skipUnless(_node_available(), 'node 없음 — JS 대조 생략')
class JsParityTests(unittest.TestCase):
"""렌더된 페이지의 미리보기 계산 == 서버 발주 계산.
손으로 옮긴 사본이 아니라 실제 페이지에서 함수 소스를 뽑아 돌린다.
어긋나면 관리자님이 화면에서 본 것과 다른 주문이 나간다.
"""
@classmethod
def setUpClass(cls):
import behive_web
cls.html = behive_web.render_html()
def _grab(self, name: str) -> str:
i = self.html.index(f'function {name}(')
depth, j, started = 0, i, False
while j < len(self.html):
if self.html[j] == '{':
depth += 1
started = True
elif self.html[j] == '}':
depth -= 1
if started and depth == 0:
return self.html[i:j + 1]
j += 1
raise AssertionError(f'{name} 추출 실패 — 함수명이 바뀌었나?')
def test_parity(self):
src = '\n'.join(self._grab(f) for f in
('tickSize', 'floorTick', 'trailCondPrice',
'allocStepQty', 'trailWeights'))
cond_cases = [{'peak': p, 'pct': c, 'floor': f}
for p in (1000, 1943, 4805, 4870, 5500, 12345, 55500, 262500, 1718000)
for c in (0.5, 1, 3, 5, 7.5, 10, 15, 20, 25, 30)
for f in (0, 4500, 5000, 250000)]
qty_cases = [{'total': q, 'w': w}
for q in (1, 2, 3, 7, 19, 100, 218, 1395, 1943)
for w in ([20, 30, 50], [30, 30, 40], [10, 20, 30, 40],
[50, 50], [100], [30, 30])]
cum_cases = [[20, 50, 100], [30, 70, 100], [25, 50, 100], [100]]
js = src + f'''
const out = {{
cond: {json.dumps(cond_cases)}.map(c => {{
const r = trailCondPrice(c.peak, c.pct, c.floor); return [r.cond, r.floorApplied]; }}),
qty: {json.dumps(qty_cases)}.map(c => allocStepQty(c.total, c.w)),
cum: {json.dumps(cum_cases)}.map(cs => trailWeights(cs.map(x => ({{cum: x}})))),
}};
console.log(JSON.stringify(out));
'''
proc = subprocess.run([shutil.which('node'), '-e', js],
capture_output=True, text=True)
self.assertEqual(proc.returncode, 0, proc.stderr)
got = json.loads(proc.stdout)
for c, (jcond, jfloor) in zip(cond_cases, got['cond']):
lv = t.compute_levels(c['peak'], c['pct'], c['floor'] or None)
self.assertEqual((lv['cond_uv'], bool(lv['floor_applied'])),
(jcond, bool(jfloor)),
f"손절선 peak={c['peak']} pct={c['pct']} floor={c['floor']}")
for c, jq in zip(qty_cases, got['qty']):
self.assertEqual(t.allocate_step_qty(c['total'], c['w']), jq,
f"수량배분 total={c['total']} w={c['w']}")
for cums, jw in zip(cum_cases, got['cum']):
raw = [{'pct': (i + 1) * 3, 'cum': v} for i, v in enumerate(cums)]
self.assertEqual([s['weight'] for s in t.normalize_steps(raw)], jw,
f'비중환산 {cums}')
if __name__ == '__main__':
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
unittest.main(verbosity=2)