"""記事「得意先別の単価・掛け率・特値はどう持つ?3通りを比較」(customer-price-master-three-ways)の計算と図。
python scripts/figs/price_master_three_ways.py (fields/sales-management-app/ で実行。外部データなし、Python 標準ライブラリだけ)
架空の卸(得意先300・商品2,000)を乱数の種を固定して作り、得意先別単価の持ち方3通りを SQLite のテーブルにして数える。
A. 得意先×商品の単価表 : customer_price(得意先, 商品, 単価) に、買う組み合わせを全部
B. 掛け率+特値 : standard_price(商品, 標準価格) + customer(得意先, 掛け率) + special(得意先, 商品, 特値)
C. 価格ランク+特値 : rank_price(ランク, 商品, 単価) + customer(得意先, ランク) + special
条件(cms/_facts.md 8章と同じ):1社が買う商品は全体の10%(200品)、特値は買う組み合わせの5%(3,000行)、ランク5段階。
掛け率はランクと同じ5つ(0.90/0.85/0.80/0.75/0.70)にして、改定前は3方式とも同じ単価になるようにした(全件で確かめる)。
単価は円未満を四捨五入(ROUND_HALF_UP。Python の round() は銀行丸めなので使わない)。
標準価格の5%改定:新しい標準価格 = 旧 × 1.05 を10円未満四捨五入。
A は「今の単価 × 1.05 を円未満四捨五入」で一括更新するのが手早いので、その結果と、B・C の「新しい標準価格 × 掛け率」を比べて、ずれる行を数える。
引き当ての順番:特値(期間内)→ 得意先別単価 → 掛け率 × 標準価格 → 標準価格。B の構成に、得意先別単価(期間なし)を少し足した形で実装し、
2026-10-16 時点で各段に当たった件数を数える。
結果は JSON で出し、記事の図(public/img/articles/customer-price-master-three-ways.svg と、本文に貼るインラインの SVG)を書き出す。
"""
import json, pathlib, random, sqlite3, sys, platform
from decimal import Decimal, ROUND_HALF_UP
ROOT = pathlib.Path(__file__).resolve().parents[2]
OUT_SVG = ROOT / 'public/img/articles/customer-price-master-three-ways.svg'
OUT_INLINE = ROOT / 'scripts/figs/.price_master_inline.html'
SEED = 20261016
C, P, USE, SPECIAL, RANKS = 300, 2000, 0.10, 0.05, 5
RATES = [Decimal('0.90'), Decimal('0.85'), Decimal('0.80'), Decimal('0.75'), Decimal('0.70')] # ランク1〜5 = 掛け率
TODAY = '2026-10-16'
def yen(x: Decimal) -> int:
return int(x.quantize(Decimal('1'), rounding=ROUND_HALF_UP))
def yen10(x: Decimal) -> int:
return int((x / 10).quantize(Decimal('1'), rounding=ROUND_HALF_UP) * 10)
rng = random.Random(SEED)
LIST = [120, 180, 250, 380, 450, 580, 720, 860, 980, 1250, 1480, 1800, 2350, 2980, 3600, 4800, 6500, 8200, 12000, 18500]
std = {p: rng.choice(LIST) for p in range(P)} # 標準価格(10円単位)
rank = {c: rng.randrange(RANKS) for c in range(C)} # 得意先のランク(0〜4)= 掛け率
per = int(P * USE) # 1社が買う品数 200
pairs = [(c, p) for c in range(C) for p in sorted(rng.sample(range(P), per))]
sp_pairs = set(rng.sample(pairs, int(len(pairs) * SPECIAL))) # 特値 3,000組
def rate_price(c, p, s=None):
return yen(Decimal((s or std)[p]) * RATES[rank[c]])
special = {}
for c, p in sorted(sp_pairs):
special[(c, p)] = yen(Decimal(rate_price(c, p)) * Decimal(rng.choice(['0.90', '0.92', '0.94', '0.96'])))
db = sqlite3.connect(':memory:')
db.executescript('''
create table customer_price(customer int, product int, price int, primary key(customer, product));
create table standard_price(product int primary key, price int);
create table customer(customer int primary key, rate real, rank int);
create table special(customer int, product int, price int, kind text, valid_from text, valid_to text);
create table rank_price(rank int, product int, price int, primary key(rank, product));
''')
db.executemany('insert into customer_price values(?,?,?)',
[(c, p, special.get((c, p), rate_price(c, p))) for c, p in pairs])
db.executemany('insert into standard_price values(?,?)', std.items())
db.executemany('insert into customer values(?,?,?)', [(c, float(RATES[r]), r) for c, r in rank.items()])
db.executemany('insert into special values(?,?,?,?,?,?)', [(c, p, v, '特値', None, None) for (c, p), v in special.items()])
db.executemany('insert into rank_price values(?,?,?)', [(r, p, yen(Decimal(s) * RATES[r])) for r in range(RANKS) for p, s in std.items()])
n = lambda sql: db.execute(sql).fetchone()[0]
rows = {
'table': n('select count(*) from customer_price'),
'rate': n('select count(*) from standard_price') + n('select count(*) from customer') + n('select count(*) from special'),
'rank': n('select count(*) from rank_price') + n('select count(*) from customer') + n('select count(*) from special'),
}
rows_detail = {
'rate': {'standard': P, 'customer': C, 'special': len(special)},
'rank': {'rankPrice': RANKS * P, 'customer': C, 'special': len(special)},
}
# 改定前、3方式の単価が全件一致するか
mis_before = 0
for c, p in pairs:
a = db.execute('select price from customer_price where customer=? and product=?', (c, p)).fetchone()[0]
b = special.get((c, p), rate_price(c, p))
r = special.get((c, p), db.execute('select price from rank_price where rank=? and product=?', (rank[c], p)).fetchone()[0])
mis_before += (a != b) or (a != r)
# 標準価格を5%改定するときに直す・見直す行
revise = {'table': rows['table'], 'rate': P + len(special), 'rank': RANKS * P + len(special)}
revise_detail = {'rate': {'standard': P, 'specialReview': len(special)}, 'rank': {'rankPrice': RANKS * P, 'specialReview': len(special)}}
# 5%改定:A は今の単価×1.05、B・C は新しい標準価格×掛け率
std_new = {p: yen10(Decimal(s) * Decimal('1.05')) for p, s in std.items()}
diff, diff_up, diff_down, maxgap = 0, 0, 0, 0
normal = [(c, p) for c, p in pairs if (c, p) not in special]
for c, p in normal:
a = yen(Decimal(rate_price(c, p)) * Decimal('1.05'))
b = rate_price(c, p, std_new)
if a != b:
diff += 1; diff_up += a > b; diff_down += a < b; maxgap = max(maxgap, abs(a - b))
# 引き当ての順番(特値→得意先別→掛け率→標準)。B の構成に、期間つきの特値と期間なしの得意先別単価を分けて持たせる
db.execute('delete from special')
sp_items = sorted(special.items())
rng2 = random.Random(SEED + 1)
n_campaign = 600
camp = {i: k for k, i in enumerate(sorted(rng2.sample(range(len(sp_items)), n_campaign)))}
recs = []
for i, ((c, p), v) in enumerate(sp_items):
if i in camp: # 期間つきの特値:半分は期間内、半分は9月で終わったもの
to = '2026-12-31' if camp[i] % 2 == 0 else '2026-09-30'
recs.append((c, p, v, '特値', '2026-07-01', to))
else:
recs.append((c, p, v, '得意先別', None, None))
db.executemany('insert into special values(?,?,?,?,?,?)', recs)
db.execute("insert into customer values(9001, null, null)") # 掛け率を決めていない新しい得意先(1社)
RESOLVE = '''
select coalesce(
(select price from special s where s.customer=:c and s.product=:p and kind='特値'
and :d between valid_from and valid_to),
(select price from special s where s.customer=:c and s.product=:p and kind='得意先別'),
(select cast(round(sp.price * cu.rate + 0.0000001) as int) from standard_price sp, customer cu
where sp.product=:p and cu.customer=:c and cu.rate is not null),
(select price from standard_price where product=:p))
'''
STEP = '''
select case
when exists(select 1 from special s where s.customer=:c and s.product=:p and kind='特値' and :d between valid_from and valid_to) then '特値'
when exists(select 1 from special s where s.customer=:c and s.product=:p and kind='得意先別') then '得意先別'
when (select rate from customer where customer=:c) is not null then '掛け率'
else '標準' end
'''
hits = {'特値': 0, '得意先別': 0, '掛け率': 0, '標準': 0}
bad = 0
for c, p in pairs:
step = db.execute(STEP, {'c': c, 'p': p, 'd': TODAY}).fetchone()[0]
hits[step] += 1
price = db.execute(RESOLVE, {'c': c, 'p': p, 'd': TODAY}).fetchone()[0]
expect = special[(c, p)] if step in ('特値', '得意先別') else rate_price(c, p)
bad += price != expect
new_cust = [db.execute(STEP, {'c': 9001, 'p': p, 'd': TODAY}).fetchone()[0] for p in range(5)]
out = {
'seed': SEED, 'run': TODAY, 'python': sys.version.split()[0], 'sqlite': sqlite3.sqlite_version, 'os': platform.platform(),
'cond': {'customers': C, 'products': P, 'useRate': USE, 'pairs': len(pairs), 'specials': len(special), 'ranks': RANKS},
'rows': rows, 'rowsDetail': rows_detail, 'mismatchBefore': mis_before,
'revise': revise, 'reviseDetail': revise_detail,
'raise5': {'normalPairs': len(normal), 'diff': diff, 'tableHigher': diff_up, 'tableLower': diff_down, 'maxGapYen': maxgap},
'resolve': {'hits': hits, 'campaignExpired': n_campaign // 2, 'resolverMismatch': bad, 'newCustomerStep': new_cust[0]},
}
print(json.dumps(out, ensure_ascii=False, indent=1))
# ---- 図(テーマ:地は白、面は #678b2c、文字は #557323、墨 #111、灰 #6b6b6b)
FONT = "'Trebuchet MS','Yu Gothic','Hiragino Sans','Meiryo',sans-serif"
G, GT, INK, MUTE, RULE, PALE = '#678b2c', '#557323', '#111111', '#6b6b6b', '#dcdcdc', '#c9d8b0'
def t(x, y, s, size=13, fill=INK, anchor='start', w=400):
return f"{s}"
def fig_rows():
W, H = 760, 400
names = [('table', '得意先×商品の単価表'), ('rate', '掛け率+特値'), ('rank', '価格ランク+特値')]
x0, x1 = 210, 540
mx = 60000
s = [f"')
return ''.join(s)
def fig_schema():
W, H = 700, 250
s = [f"')
return ''.join(s)
def fig_resolve():
W, H = 700, 200
steps = [('特値(期間内)', hits['特値']), ('得意先別単価', hits['得意先別']), ('掛け率×標準価格', hits['掛け率']), ('標準価格', hits['標準'])]
s = [f"')
return ''.join(s)
OUT_SVG.parent.mkdir(parents=True, exist_ok=True)
OUT_SVG.write_text(fig_rows(), encoding='utf-8')
OUT_INLINE.write_text(fig_schema() + '\n\n' + fig_resolve() + '\n', encoding='utf-8')