監査に強い暗号資産台帳：設計・SQL・検証

スキーマ最小セット（推奨）

1. 事実（Fact）

addresses(address, entity_id, label, is_internal BOOL, valid_from, valid_to)

tx(tx_id, chain, block_time, from_addr, to_addr, tx_type, hash, raw JSON)

transfers(tx_id, leg_id, asset, amount, from_addr, to_addr, fee_asset, fee_amount)

rates(as_of_time, asset, base_ccy, rate, source)

2. 解釈（Rules）

classification_rules(rule_id, predicate_sql, accounting_event, priority)

例: 「to_addr が自社 && from_addr も自社 → internal_transfer」

valuation_policy(asset, method ENUM('fifo','wac','spot'), base_ccy)

3. 記録（Ledger）

journal_entries(entry_id, tx_id, journal_no, line_no, account, asset, qty, amount_base, memo)

journal_no 単位で総額=0になるのが不変条件。

audit_log(seq BIGINT, ts, actor, action, payload, prev_hash, hash)

不変条件（Invariant）

SUM(amount_base) OVER (PARTITION BY journal_no) = 0

transfers.tx_id は journal_entries.tx_id に完全カバーされる

同一入力（raw events + rules + rates）から同一 journal_entries が再現される（ルールはバージョン固定）

仕訳生成の要点（3ケース）
ケースA：内部移動（自社 → 自社）

PL影響なし。資産内振替のみ。

手数料があれば費用認識（Gas Expense など）。

ケースB：外部流出（自社 → 外部）

目的に応じて売却／支払／出庫に分類。

評価差額の扱いは方針（FIFO/WAC/スポット）に従う。

ケースC：外部流入（外部 → 自社）

原因により売上／負債／入庫。未判別はサスペンスで隔離し、後続ワークフローで消し込み。

SQL（BigQuery）例：内部移動＋手数料
-- 1) 内部先・内部元の判定
WITH internal_flags AS (
SELECT
t.tx_id, t.leg_id, t.asset, t.amount,
t.from_addr, t.to_addr,
af.is_internal AS from_internal,
at.is_internal AS to_internal,
t.fee_asset, t.fee_amount, x.block_time
FROM transfers t
JOIN tx x USING (tx_id)
LEFT JOIN addresses af ON t.from_addr = af.address AND (af.valid_from IS NULL OR x.block_time >= af.valid_from) AND (af.valid_to IS NULL OR x.block_time < af.valid_to)
LEFT JOIN addresses at ON t.to_addr = at.address AND (at.valid_from IS NULL OR x.block_time >= at.valid_from) AND (at.valid_to IS NULL OR x.block_time < at.valid_to)
),
-- 2) 内部移動の抽出
internal_moves AS (
SELECT * FROM internal_flags
WHERE from_internal AND to_internal
),
-- 3) スポット評価（例：即時スポット。方針に合わせてFIFO/WACに置換）
priced AS (
SELECT
i.*,
r.rate AS spot_rate, -- asset -> base_ccy
rf.rate AS fee_rate,
TIMESTAMP_TRUNC(block_time, HOUR) AS rate_ts
FROM internal_moves i
LEFT JOIN rates r ON r.asset = i.asset AND r.base_ccy = 'USD' AND r.as_of_time = TIMESTAMP_TRUNC(i.block_time, HOUR)
LEFT JOIN rates rf ON rf.asset = i.fee_asset AND rf.base_ccy = 'USD' AND rf.as_of_time = TIMESTAMP_TRUNC(i.block_time, HOUR)
),
-- 4) 二重仕訳の生成（journal_no = tx_id）
journal AS (
SELECT
tx_id AS journal_no,
1 AS line_no,
'Crypto Assets:' || asset AS account,
asset,
amount AS qty,
amount * spot_rate AS amount_base,
'internal transfer in' AS memo
FROM priced
UNION ALL
SELECT
tx_id, 2,
'Crypto Assets:' || asset,
asset,
-amount,
-amount * spot_rate,
'internal transfer out'
FROM priced
UNION ALL
SELECT
tx_id, 3,
'Gas Expense',
fee_asset,
NULL,
fee_amount * fee_rate,
'network fee'
FROM priced
WHERE fee_amount IS NOT NULL AND fee_amount > 0
UNION ALL
SELECT
tx_id, 4,
'Crypto Assets:' || fee_asset,
fee_asset,
NULL,
- (fee_amount * fee_rate),
'fee payment'
FROM priced
WHERE fee_amount IS NOT NULL AND fee_amount > 0
)
SELECT * FROM journal
ORDER BY journal_no, line_no;

チェック：journal_no ごとの SUM(amount_base)=0。手数料がある場合も4行で釣り合うはずです。

dbt 統合（変換とテスト）

models/ledger/journal_internal.sql

{{ config(materialized='table') }}
SELECT * FROM {{ ref('journal_internal_generation') }}

models/ledger/schema.yml

version: 2
models:

• name: journal_internal
  tests:
  • dbt_utils.expression_is_true:
    expression: "ABS(SUM(amount_base)) < 0.0001"
    group_by: ["journal_no"]
  • relationships:
    to: ref('transfers')
    field: tx_id
    to_field: tx_id

目的はビルド時に壊すこと。壊れたら直す。監査の前に落とす。

データ品質：Great Expectations 例

expectations/journal_checks.py

from great_expectations.dataset import PandasDataset

class JournalDataset(PandasDataset):
_expectation_suite_name = "journal_suite"

例：ジャーナルごとゼロサム

def expect_zero_sum_per_journal(df):
s = df.groupby("journal_no")["amount_base"].sum().abs().max()
assert s < 1e-4, f"Non-zero journal detected: max abs sum = {s}"

CI では dbt の run + test の後にこのスクリプトを走らせます。

変更不能な監査ログ（Hash-chained）
-- 新規イベントの挿入時にチェーンする（簡略版）
INSERT INTO audit_log (seq, ts, actor, action, payload, prev_hash, hash)
WITH last AS (
SELECT COALESCE(MAX(seq), 0) AS last_seq, ANY_VALUE(hash) AS last_hash FROM audit_log
),
ins AS (
SELECT
(SELECT last_seq FROM last) + 1 AS seq,
CURRENT_TIMESTAMP() AS ts,
'system' AS actor,
'journal_materialize' AS action,
TO_JSON_STRING(@payload) AS payload,
(SELECT last_hash FROM last) AS prev_hash
)
SELECT
seq, ts, actor, action, payload, prev_hash,
TO_HEX(SHA256(CONCAT(CAST(seq AS STRING), CAST(ts AS STRING), actor, action, payload, IFNULL(prev_hash,''))))
FROM ins;

ポイント

audit_log は append-only、UPDATE/DELETE は禁止。

ハッシュは前レコードのハッシュを含める（改ざん検知）。

変更系は**署名（サービスアカウント／人）**を記録。

AML ルールの最小実装

rules/aml.yml

• id: sanction_match
  type: list_match
  on: ["from_addr","to_addr"]
  list: "sanctions.csv"
  action: "block_and_alert"
• id: high_velocity_outflow
  type: velocity
  predicate: "entity_id = 'corp'"
  window_hours: 24
  threshold_usd: 500000
  action: "review"
• id: unusual_route
  type: path
  max_hops: 4
  denylist_tags: ["mixer","darknet_service"]
  action: "review"

SQL 化の基本は「明示的に危険を定義し、JOIN して抽出」。誤検知は抑止より説明を優先。

指標と運用

オンチェーン照合率：transfers とチェーン照会の一致率（>99%を目標）。

監査指摘件数：四半期あたり運用改善で継続減。

TTV（Time to Voucher）：原始データ→仕訳までの平均所要時間。

運用チェックリスト

事実テーブルはイミュータブル運用（修正は差分テーブル）

ルールはバージョン管理（rule_version）

評価レートはソース・時刻固定

すべての成果物に生成ハッシュ/署名付与

CI に dbt + GE + pytest を束ねる

よくある質問（短問短答）

Q. どこから始めれば良い？
A. 最初は内部移動と手数料だけを自動化し、ゼロサム検証をCIに入れる。

Q. FIFO/WAC/スポットの選定は？
A. 会計方針に依存。コードは戦略パターンにして差し替え可能に。

Q. 取引所CSVのズレ対策は？
A. ハッシュ合意（原本のSHA256）と差分取り込み。後勝ちではなく再計算で整合性担保。

Q. 監査ログは重くない？
A. 書き込みは軽量。集計は別系（BIにレプリカ）で運用する。

まとめ一言

数字は出発点、説明責任は目的地。まずは小さく、早く動かし、証跡で支えましょう。

著者：近藤 洋士（LedgerField）
※本記事は一般的情報であり、法務・会計・税務の助言ではありません。

https://ledgerfield.tokyo/