kjh2064
f99f9821d2
settings/account_snapshot SQLite를 직접 편집하는 잠금/승인/변경이력 기반 웹 에디터를 추가하고, 2026-06-21 비판적 리뷰에서 요청된 테이블별 그리드 조회 기능(Tabler CDN)을 /tables 경로로 덧붙인다. - 잠금(lock)·승인(approval)·undo·변경로그 전체 감사 추적 - KIS Collection 대시보드 통합(별도 SQLite, 워크스페이스 DB와 분리) - WBS-7.10: 워크스페이스/KIS수집/정성매도전략 3개 SQLite, 11개 테이블을 Tabler 그리드로 조회 — 테이블명은 고정 화이트리스트와 정확히 일치할 때만 SQL에 사용(SQL 인젝션 방지, 단위테스트로 검증)
250 lines
8.9 KiB
Python
250 lines
8.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from src.quant_engine.snapshot_admin_server_v1 import build_ui_state
|
|
from src.quant_engine.snapshot_admin_store_v1 import (
|
|
ACCOUNT_SNAPSHOT_CANONICAL_COLUMNS,
|
|
export_payload,
|
|
import_seed_json,
|
|
load_approval_for_domain,
|
|
load_change_log_rows,
|
|
load_locks,
|
|
load_account_snapshot_rows,
|
|
load_settings_rows,
|
|
parse_account_snapshot_tsv,
|
|
open_connection,
|
|
lock_conflicts_for_rows,
|
|
validate_account_snapshot_rows,
|
|
validate_settings_rows,
|
|
build_validation_suggestions,
|
|
build_safe_autofix_actions,
|
|
apply_safe_autofix_action,
|
|
set_lock,
|
|
undo_last_change,
|
|
write_export_json,
|
|
)
|
|
|
|
|
|
def _seed_json(path: Path) -> None:
|
|
payload = {
|
|
"data": {
|
|
"settings": {
|
|
"total_asset_krw": 150000000,
|
|
"weekly_target_cash_pct": 14,
|
|
"orbit_start_yyyymm": "2026-01",
|
|
},
|
|
"account_snapshot": [
|
|
{
|
|
"captured_at": "2026-06-21T09:00:00+09:00",
|
|
"account": "real",
|
|
"account_type": "일반계좌",
|
|
"ticker": "005930",
|
|
"name": "삼성전자",
|
|
"holding_quantity": 10,
|
|
"available_quantity": 10,
|
|
"average_cost": 70000,
|
|
"total_cost": 700000,
|
|
"current_price": 71000,
|
|
"market_value": 710000,
|
|
"profit_loss": 10000,
|
|
"return_pct": 1.43,
|
|
"immediate_cash": 1000000,
|
|
"settlement_cash_d2": 1000000,
|
|
"available_cash": 1000000,
|
|
"open_order_amount": 0,
|
|
"monthly_contribution_limit": "",
|
|
"monthly_contribution_used": "",
|
|
"parse_status": "CAPTURE_READ_OK",
|
|
"user_confirmed": "Y",
|
|
"stop_price": 65000,
|
|
"highest_price_since_entry": 72000,
|
|
"entry_date": "2026-06-01",
|
|
"entry_stage": "stage_1",
|
|
"position_type": "core",
|
|
"last_updated": "2026-06-21T09:05:00+09:00",
|
|
}
|
|
],
|
|
}
|
|
}
|
|
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
|
|
def test_seed_import_and_export_round_trip(tmp_path):
|
|
db_path = tmp_path / "snapshot.db"
|
|
seed_path = tmp_path / "seed.json"
|
|
_seed_json(seed_path)
|
|
|
|
summary = import_seed_json(db_path, seed_path)
|
|
assert summary["settings_rows"] == 3
|
|
assert summary["account_snapshot_rows"] == 1
|
|
|
|
settings_rows = load_settings_rows(db_path)
|
|
assert settings_rows[0]["key"] == "total_asset_krw"
|
|
assert settings_rows[0]["value"] == 150000000
|
|
|
|
snapshot_rows = load_account_snapshot_rows(db_path)
|
|
assert snapshot_rows[0]["ticker"] == "005930"
|
|
assert snapshot_rows[0]["parse_status"] == "CAPTURE_READ_OK"
|
|
|
|
exported = export_payload(db_path)
|
|
assert exported["data"]["settings"]["weekly_target_cash_pct"] == 14
|
|
assert exported["data"]["account_snapshot"][0]["name"] == "삼성전자"
|
|
|
|
out = write_export_json(db_path, tmp_path / "export.json")
|
|
assert out.exists()
|
|
|
|
|
|
def test_parse_account_snapshot_tsv_supports_headerless_and_header_rows():
|
|
headerless = "\n".join(
|
|
[
|
|
"\t".join(ACCOUNT_SNAPSHOT_CANONICAL_COLUMNS),
|
|
"\t".join(
|
|
[
|
|
"2026-06-21T09:00:00+09:00",
|
|
"real",
|
|
"일반계좌",
|
|
"005930",
|
|
"삼성전자",
|
|
"10",
|
|
"10",
|
|
"70000",
|
|
"700000",
|
|
"71000",
|
|
"710000",
|
|
"10000",
|
|
"1.43",
|
|
"1000000",
|
|
"1000000",
|
|
"1000000",
|
|
"0",
|
|
"",
|
|
"",
|
|
"CAPTURE_READ_OK",
|
|
"Y",
|
|
"65000",
|
|
"72000",
|
|
"2026-06-01",
|
|
"stage_1",
|
|
"core",
|
|
"2026-06-21T09:05:00+09:00",
|
|
]
|
|
),
|
|
]
|
|
)
|
|
rows = parse_account_snapshot_tsv(headerless)
|
|
assert rows[0]["ticker"] == "005930"
|
|
assert rows[0]["holding_quantity"] == 10
|
|
|
|
with_header = "captured_at\taccount\tticker\n2026-06-21T09:00:00+09:00\treal\t005930"
|
|
rows2 = parse_account_snapshot_tsv(with_header)
|
|
assert rows2[0]["account"] == "real"
|
|
assert rows2[0]["ticker"] == "005930"
|
|
|
|
|
|
def test_build_ui_state_reports_schema(tmp_path):
|
|
db_path = tmp_path / "snapshot.db"
|
|
seed_path = tmp_path / "seed.json"
|
|
_seed_json(seed_path)
|
|
import_seed_json(db_path, seed_path)
|
|
|
|
state = build_ui_state(db_path)
|
|
assert state["summary"]["settings_rows"] == 3
|
|
assert state["account_snapshot_columns"][: len(ACCOUNT_SNAPSHOT_CANONICAL_COLUMNS)] == ACCOUNT_SNAPSHOT_CANONICAL_COLUMNS
|
|
|
|
|
|
def test_change_log_approval_and_lock_workflow(tmp_path):
|
|
db_path = tmp_path / "snapshot.db"
|
|
seed_path = tmp_path / "seed.json"
|
|
_seed_json(seed_path)
|
|
import_seed_json(db_path, seed_path)
|
|
|
|
with open_connection(db_path) as conn:
|
|
set_lock(conn, "settings", "*", locked_by="tester", reason="review")
|
|
conn.commit()
|
|
|
|
locks = load_locks(db_path)
|
|
assert locks and locks[0]["domain"] == "settings"
|
|
|
|
approval = load_approval_for_domain(db_path, "settings")
|
|
assert approval["status"] == "PENDING"
|
|
|
|
changes = load_change_log_rows(db_path, limit=10)
|
|
assert changes
|
|
|
|
|
|
def test_lock_conflicts_detect_row_targets(tmp_path):
|
|
db_path = tmp_path / "snapshot.db"
|
|
seed_path = tmp_path / "seed.json"
|
|
_seed_json(seed_path)
|
|
import_seed_json(db_path, seed_path)
|
|
|
|
with open_connection(db_path) as conn:
|
|
set_lock(conn, "settings", "total_asset_krw", locked_by="tester", reason="review")
|
|
set_lock(conn, "account_snapshot", "005930", locked_by="tester", reason="review")
|
|
conn.commit()
|
|
|
|
settings_conflicts = lock_conflicts_for_rows(
|
|
db_path,
|
|
"settings",
|
|
[{"key": "total_asset_krw", "value": 123, "note": ""}],
|
|
)
|
|
snapshot_conflicts = lock_conflicts_for_rows(
|
|
db_path,
|
|
"account_snapshot",
|
|
[{"ticker": "005930", "name": "삼성전자", "ordinal": 1}],
|
|
)
|
|
|
|
assert settings_conflicts and settings_conflicts[0]["target_ref"] == "total_asset_krw"
|
|
assert snapshot_conflicts and snapshot_conflicts[0]["target_ref"] == "005930"
|
|
|
|
|
|
def test_undo_last_change_restores_previous_snapshot(tmp_path):
|
|
db_path = tmp_path / "snapshot.db"
|
|
seed_path = tmp_path / "seed.json"
|
|
_seed_json(seed_path)
|
|
import_seed_json(db_path, seed_path)
|
|
|
|
with open_connection(db_path) as conn:
|
|
from src.quant_engine.snapshot_admin_store_v1 import replace_settings
|
|
|
|
replace_settings(conn, [{"ordinal": 1, "key": "total_asset_krw", "value": 123, "note": "edited"}])
|
|
|
|
with open_connection(db_path) as conn:
|
|
undo_last_change(conn, "settings")
|
|
|
|
settings_rows = load_settings_rows(db_path)
|
|
assert settings_rows[0]["value"] == 150000000
|
|
|
|
|
|
def test_validation_helpers_detect_invalid_rows():
|
|
assert "settings.total_asset_krw is required" in validate_settings_rows([{"key": "weekly_target_cash_pct", "value": 10}])
|
|
assert "account_snapshot row 1: ticker required" in validate_account_snapshot_rows(
|
|
[{"captured_at": "2026-06-21", "account": "real", "name": "삼성전자", "parse_status": "BAD"}]
|
|
)
|
|
suggestions = build_validation_suggestions(
|
|
[{"key": "weekly_target_cash_pct", "value": 10}],
|
|
[{"captured_at": "2026-06-21", "account": "real", "account_type": "일반계좌", "ticker": "005930", "name": "삼성전자", "parse_status": "CAPTURE_READ_OK", "user_confirmed": "N"}],
|
|
)
|
|
assert any("user_confirmed=Y" in item for item in suggestions)
|
|
actions = build_safe_autofix_actions(
|
|
[{"key": "total_asset_krw", "value": 150000000}],
|
|
[{"captured_at": "2026-06-21", "account": "real", "account_type": "일반계좌", "ticker": "005930", "name": "삼성전자", "parse_status": "CAPTURE_READ_OK", "user_confirmed": "N", "entry_stage": "stage_1", "position_type": ""}],
|
|
)
|
|
assert any(item["action_id"] == "confirm_captured_rows" for item in actions)
|
|
|
|
|
|
def test_safe_autofix_updates_snapshot_defaults(tmp_path):
|
|
db_path = tmp_path / "snapshot.db"
|
|
seed_path = tmp_path / "seed.json"
|
|
_seed_json(seed_path)
|
|
import_seed_json(db_path, seed_path)
|
|
|
|
with open_connection(db_path) as conn:
|
|
result = apply_safe_autofix_action(conn, "confirm_captured_rows")
|
|
assert result["status"] == "AUTOFIXED"
|
|
|
|
snapshot_rows = load_account_snapshot_rows(db_path)
|
|
assert all(row.get("user_confirmed") == "Y" or str(row.get("parse_status")) != "CAPTURE_READ_OK" for row in snapshot_rows)
|