Python으로 매출 증감을 가격·수량·신규·중단 품목으로 나누기
전기 매출이 2,300이고 당기가 2,320이면 총증감은 20입니다. 이 숫자만으로는 같은 품목의 단가가 바뀌었는지, 판매 수량이 달라졌는지, 한 기간에만 나타난 품목이 있는지 알기 어렵습니다. 두 기간의 SKU를 연결한 뒤 가격 -100, 수량 +20, 신규 +400, 중단 -300으로 나누면 합계가 다시 +20이 되는지 확인할 수 있습니다.
이 글은 Excel 등에서 준비한 두 개의 CSV를 Python 3.12 표준 라이브러리로 계산하는 실습입니다. 가상 품목 A·B·C·D를 사용하며 실제 거래나 회계 제도, 성과평가를 검증하는 사례는 아닙니다. 실제 Excel 앱은 실행하지 않았습니다. 결과 CSV와 정책 기록을 만들어, 어떤 계산 순서와 입력 기준으로 증감이 연결됐는지 살펴봅니다.
먼저 정할 기준: 한 SKU, 한 기간, 한 단가
각 기간 CSV에는 SKU별 한 행을 둡니다. 품목명보다 기간 사이에 변하지 않는 SKU가 필요합니다. 같은 제품의 코드가 바뀌었다면 프로그램은 옛 키의 중단과 새 키의 신규로 분류하므로, 코드 변경 관계를 먼저 확인해야 합니다.
이 실습에서 ‘신규’와 ‘중단’은 두 입력 파일의 키 존재 여부를 나타냅니다. 당기에만 있는 키가 신규, 전기에만 있는 키가 중단입니다. 실제 출시일·판매종료일을 증명하는 분류가 아닙니다. 자료 누락 때문에 한 파일에서 빠진 품목을 진짜 중단으로 해석하지 마세요.
원본이 거래별 자료라 같은 SKU가 여러 번 나온다면, 먼저 어떤 단가를 쓸지 정해야 합니다. 단가가 모두 같은 거래의 수량을 합하는 경우와 여러 단가의 가중평균을 만드는 경우는 준비 과정이 다릅니다. 이 스크립트는 중복 SKU를 자동 합산하거나 첫 행만 남기지 않고 오류로 멈춥니다. 평균 단가를 두 자리로 반올림해 넣으면 원래 거래 매출과 차이가 생길 수 있으므로 원본 매출의 별도 대조가 필요합니다.
합성 입력 두 파일과 고정 기대값
두 파일의 헤더는 sku,quantity,unit_price,currency,status입니다. 실습 금액은 같은 KRW 표기로 계산하되 출력에서 두 소수자리를 유지합니다. 소수 표시는 계산 예제의 형식이며 실제 원화 정산 단위를 정하는 규칙이 아닙니다.
전기 input/previous.csv:
sku,quantity,unit_price,currency,status
A,10,100,KRW,posted
B,20,50,KRW,posted
D,4,75,KRW,posted
당기 input/current.csv:
sku,quantity,unit_price,currency,status
A,12,110,KRW,posted
B,15,40,KRW,posted
C,5,80,KRW,posted
전기는 A의 1,000+B의 1,000+D의 300=2,300, 당기는 A의 1,320+B의 600+C의 400=2,320입니다. A·B는 공통 품목, C는 당기에만 있는 키, D는 전기에만 있는 키입니다. 데이터에 취소 상태나 반품 음수는 없습니다.
가격을 먼저, 수량을 나중에 바꾸기
전기 수량과 단가를 q0·p0, 당기를 q1·p1이라고 두고 다음 순서를 고정합니다.
- 전기 매출 q0×p0에서 수량 q0를 유지한 채 단가만 p1로 바꿉니다.
- 이어서 당기 단가 p1을 유지한 채 수량을 q1로 바꿉니다.
가격효과 = q0 × (p1 - p0)
수량효과 = p1 × (q1 - q0)
가격효과 + 수량효과 = q1×p1 - q0×p0
A는 10×100=1,000에서 10×110=1,100을 거쳐 12×110=1,320이 됩니다. 가격효과는 +100, 수량효과는 +220입니다. B는 20×50=1,000에서 20×40=800을 거쳐 15×40=600이 되므로 두 효과가 각각 -200입니다.
순서를 달리하면 개별 효과도 달라집니다. A의 수량을 먼저 바꾸면 수량효과는 100×(12-10)=200, 가격효과는 12×(110-100)=120입니다. 합은 둘 다 320이지만 배분이 다릅니다. 아래 코드는 가격 먼저라는 기준을 정책 파일에 남깁니다. 산술 분해를 실제 원인의 크기라고 단정하지 않습니다.
신규·중단까지 포함한 연결표
| SKU | 구분 | 전기 매출 | 당기 매출 | 가격효과 | 수량효과 | 신규효과 | 중단효과 | 증감 |
|---|---|---|---|---|---|---|---|---|
| A | COMMON | 1000.00 | 1320.00 | 100.00 | 220.00 | 0.00 | 0.00 | 320.00 |
| B | COMMON | 1000.00 | 600.00 | -200.00 | -200.00 | 0.00 | 0.00 | -400.00 |
| C | NEW | 0.00 | 400.00 | 0.00 | 0.00 | 400.00 | 0.00 | 400.00 |
| D | DISCONTINUED | 300.00 | 0.00 | 0.00 | 0.00 | 0.00 | -300.00 | -300.00 |
| 합계 | 2300.00 | 2320.00 | -100.00 | 20.00 | 400.00 | -300.00 | 20.00 |
각 행에서 네 효과의 합을 그 SKU의 증감과 비교하고, 전체 합계에서도 같은 대조를 합니다. unexplained가 모두 0.00이면 계산 기준 안에서 남은 산술 차이가 없습니다. C의 전기 수량·단가, D의 당기 수량·단가는 결과 CSV에서 빈칸으로 둡니다. 키가 없다는 사실을 관측 수량 0과 구분하기 위한 표시입니다. 그 기간의 매출 기여는 0.00으로 계산합니다.
파일을 내려받아 Python에서 실행하기
매출 증감 연결표 실습 ZIP 다운로드에는 스크립트, 정상·오류 입력 10개, 기대 출력, 자동 시험과 한국어 README가 들어 있습니다. ZIP을 새 폴더에 풀고 sales_bridge.py가 있는 폴더에서 터미널을 엽니다. 별도 pip 패키지는 필요 없습니다.
Python이 없다면 Python 공식 다운로드에서 운영체제에 맞는 설치 안내를 확인합니다. 다음 명령으로 실제 실행 버전을 먼저 확인합니다.
python --version
Windows에서는 환경에 따라 python 대신 py -3.12, macOS·Linux에서는 python3를 쓸 수 있습니다. 아래 모든 명령의 첫 단어를 같은 실행기로 바꿉니다. 실제 제작 검증은 Linux의 CPython 3.12.14에서 했습니다. 터미널의 현재 폴더가 다르면 입력 경로를 찾지 못하므로, 먼저 압축을 푼 실습 폴더로 이동합니다.
python sales_bridge.py --previous input/previous.csv --current input/current.csv --currency KRW --output-dir result_main
정상 입력의 종료코드는 0이고 다음 두 줄이 출력됩니다.
OK: previous=2300.00 current=2320.00 change=20.00
effects=-100.00,20.00,400.00,-300.00 unexplained=0.00
result_main에 bridge_rows.csv, bridge_totals.csv, run_policy.json 세 파일이 생깁니다. 첫 파일은 SKU별 계산, 두 번째는 한 행의 전체 합계입니다. 정책 파일에는 가격 먼저·수량 나중 식, 실행 버전, 입력 행 수, 입력 SHA-256이 남습니다. expected 폴더의 기준 CSV와 헤더·행·금액을 비교하세요.
이미 같은 이름의 결과 폴더가 있으면 오류로 멈춥니다. 다시 실행할 때는 result_main2처럼 새 이름을 지정합니다. 원본 입력은 읽기 전용으로 사용하며, 입력 검증과 계산이 끝난 뒤 별도 임시 폴더에서 결과를 준비합니다. 같은 결과 폴더로 여러 프로세스를 동시에 실행하는 사용은 실습 범위에서 제외합니다.
0과 키 부재를 따로 시험하기
전기 수량 0은 신규가 아닙니다
zero_previous.csv에는 Z,0,100이 있고 zero_current.csv에는 Z,2,110이 있습니다. 두 파일 모두에 키가 있으므로 공통 품목입니다.
python sales_bridge.py --previous input/zero_previous.csv --current input/zero_current.csv --currency KRW --output-dir result_zero
기대값은 COMMON, 가격효과 0.00, 수량효과 220.00, 신규효과 0.00입니다. 전기 수량이 0이어서 증감률의 분모가 0이 되는 상황에서도 금액 연결표는 계산할 수 있습니다. 이 예제는 변화율을 계산하지 않습니다.
당기 수량 0은 키의 중단과 다릅니다
decimal_previous.csv는 Z,1,0.10, decimal_current.csv는 Z,0,0.20입니다. 십진 문자열의 계산을 확인하기 위한 소수 단가 반례이며 실제 원화 결제 단위의 적용 사례는 아닙니다.
python sales_bridge.py --previous input/decimal_previous.csv --current input/decimal_current.csv --currency KRW --output-dir result_decimal
기대값은 COMMON, 가격효과 +0.10, 수량효과 -0.20, 증감 -0.10, 중단효과 0.00입니다. 수량이 0이라는 이유로 Z를 입력에서 삭제하면 분해 결과가 달라지므로 원래 관측값을 보존합니다.
키가 정말 없을 때는 신규로 연결됩니다
empty.csv는 헤더만 있는 유효한 빈 기간입니다. 전기를 이 파일로 바꿔 당기의 세 품목과 비교합니다.
python sales_bridge.py --previous input/empty.csv --current input/current.csv --currency KRW --output-dir result_new
전기 0.00, 당기 2320.00이며 신규효과 2320.00입니다. 빈 수량 셀을 0으로 바꾸는 처리와 헤더만 있는 완전한 빈 기간은 서로 다른 입력입니다.
입력 오류를 먼저 막고 복구하기
CSV는 UTF-8 또는 UTF-8 BOM, 쉼표 구분, 고정 헤더·열 순서입니다. 파일당 최대 2MiB·기간별 최대 10,000행, 수량 0~1,000,000,000의 정수, 단가 정수부 최대 9자리·소수부 최대 2자리로 제한합니다. 숫자 앞뒤 공백, 선행 0 수량, 음수·지수·NaN·Infinity·천 단위 쉼표는 받지 않습니다. 단가와 수량 0은 허용합니다. SKU는 영문으로 시작하는 영문·숫자·밑줄·하이픈 1~32자이고 대소문자를 구분합니다.
currency는 실행에서 지정한 대문자 세 글자 코드와 같아야 하며 다른 통화를 환산하지 않습니다. status는 posted만 허용합니다. 취소 행을 몰래 제외하거나 음수 반품을 양수로 바꾸지 않습니다. 취소·반품을 포함하는 자료는 담당자가 승인한 집계와 단가 정책을 별도로 정해야 합니다.
다음 세 명령은 각각 중복 키, 음수 수량, 취소 상태를 재현합니다.
python sales_bridge.py --previous input/bad_duplicate.csv --current input/current.csv --currency KRW --output-dir result_bad_duplicate
python sales_bridge.py --previous input/bad_negative.csv --current input/current.csv --currency KRW --output-dir result_bad_negative
python sales_bridge.py --previous input/bad_cancelled.csv --current input/current.csv --currency KRW --output-dir result_bad_cancelled
기대 종료코드는 모두 2입니다. 각각 DUPLICATE_SKU·QUANTITY·STATUS가 표시되고 해당 결과 폴더는 생기지 않습니다.
| 오류 | 먼저 확인할 것 | 복구할 때 지킬 기준 |
|---|---|---|
| DUPLICATE_SKU | 중복 추출인지 여러 거래인지 | 승인된 기간별 한 행 기준을 마련하고 재실행, 임의 첫 행 선택 금지 |
| QUANTITY·PRICE | 빈칸·음수·단위·소수 표기 | 원문과 의도를 확인, 빈칸 0 처리·반품 부호 변경 금지 |
| STATUS | 취소·반품 등 포함 정책 | posted만 쓸 수 있는 집계인지 확인, 자동 삭제 금지 |
| CURRENCY | 두 파일의 통화 | 같은 통화끼리 분리, 추정 환율 사용 금지 |
| HEADER·WIDTH | 헤더·열 순서·쉼표 CSV | 사본에서 열 구조 수정, 공백 행도 확인 |
| FILE_LIMIT | 파일당 2MiB 초과 | 작은 검증용 사본 준비, 연결 기준 없이 기간을 임의 분할하지 않기 |
| OUTPUT_EXISTS | 이전 결과 폴더 | 새 폴더 이름 사용 |
| OUTPUT_PARENT | 결과 상위 폴더 | 있는 상위 폴더 아래에 새 이름 지정 |
| INPUT_CHANGED | 실행 중 수정 여부 | 사본을 고정하고 수정이 없는 상태에서 재실행 |
오류 메시지에는 입력 파일과 레코드 번호, 물리 줄 번호가 함께 표시됩니다. 파일을 UTF-8이 아닌 다른 인코딩으로 저장했거나 인용부호 구조가 깨졌다면 읽기 단계에서 종료합니다. 실제 원본을 덮어쓰며 고치기보다 사본에서 입력 계약을 맞추세요.
전체 실행 코드
단가는 CSV 문자열에서 직접 Decimal로 읽습니다. float를 거치지 않습니다. Python Decimal 공식 문서는 십진수 표현, 연산 정밀도와 부정확한 연산의 예외 처리를 설명합니다. 여기서는 자릿수·행 수를 제한하고 정밀도 50을 사용하며 금액을 중간에 반올림하지 않습니다. 실습의 두 소수자리 출력이 현업의 회계·통화 규칙을 대신하지 않습니다.
Python csv 공식 문서에 맞춰 파일을 newline=''로 열고, 읽은 문자열의 타입과 표기를 직접 검사합니다. 아래 코드는 ZIP의 sales_bridge.py와 같습니다.
"""Synthetic two-period sales bridge. Python 3.12, standard library only."""
import argparse
import csv
from decimal import Decimal, Inexact, localcontext
import hashlib
import io
import json
from pathlib import Path
import re
import shutil
import sys
import tempfile
HEADER = ['sku', 'quantity', 'unit_price', 'currency', 'status']
AMOUNTS = ['previous_revenue', 'current_revenue', 'price_effect',
'quantity_effect', 'new_effect', 'discontinued_effect', 'change']
ROW_FIELDS = ['sku', 'classification', 'previous_quantity', 'previous_price',
'current_quantity', 'current_price', 'currency', *AMOUNTS,
'unexplained']
TOTAL_FIELDS = ['currency', *AMOUNTS, 'unexplained']
ZERO = Decimal('0')
MAX_BYTES = 2 * 1024 * 1024
def read_snapshot(path):
with Path(path).open('rb') as f:
raw = f.read(MAX_BYTES + 1)
if len(raw) > MAX_BYTES:
raise ValueError(f'FILE_LIMIT: {path}: maximum 2MiB')
return raw
def read_period(path, currency, raw=None):
result = {}
if raw is None:
raw = read_snapshot(path)
with io.StringIO(raw.decode('utf-8-sig'), newline='') as f:
reader = csv.reader(f, strict=True)
if next(reader, None) != HEADER:
raise ValueError(f'HEADER: {path}: expected {HEADER}')
for number, fields in enumerate(reader, 1):
where = f'{path}: record {number} (physical line {reader.line_num})'
if number > 10000:
raise ValueError(f'ROW_LIMIT: {where}: maximum 10000 records')
if len(fields) != len(HEADER):
raise ValueError(f'WIDTH: {where}: expected 5 fields')
sku, quantity, price, row_currency, status = fields
if not re.fullmatch(r'[A-Za-z][A-Za-z0-9_-]{0,31}', sku):
raise ValueError(f'SKU: {where}: use an ASCII code, length 1..32')
if sku in result:
raise ValueError(f'DUPLICATE_SKU: {where}: {sku}')
if (not re.fullmatch(r'0|[1-9][0-9]{0,9}', quantity)
or int(quantity) > 1000000000):
raise ValueError(f'QUANTITY: {where}: integer 0..1000000000')
if not re.fullmatch(r'(0|[1-9][0-9]{0,8})(\.[0-9]{1,2})?', price):
raise ValueError(f'PRICE: {where}: nonnegative, at most 9+2 digits')
if row_currency != currency:
raise ValueError(f'CURRENCY: {where}: expected {currency}')
if status != 'posted':
raise ValueError(f'STATUS: {where}: only posted is supported')
result[sku] = (int(quantity), Decimal(price))
return result
def amount(value):
return format(ZERO if value == 0 else value, '.2f')
def calculate(previous, current, currency):
rows = []
totals = {key: ZERO for key in AMOUNTS}
with localcontext() as context:
context.prec = 50
context.traps[Inexact] = True
for sku in sorted(previous.keys() | current.keys()):
q0, p0 = previous.get(sku, (0, ZERO))
q1, p1 = current.get(sku, (0, ZERO))
r0, r1 = q0 * p0, q1 * p1
price = quantity = new = discontinued = ZERO
if sku in previous and sku in current:
classification = 'COMMON'
price = q0 * (p1 - p0)
quantity = p1 * (q1 - q0)
elif sku in current:
classification = 'NEW'
new = r1
else:
classification = 'DISCONTINUED'
discontinued = -r0
change = r1 - r0
unexplained = change - (price + quantity + new + discontinued)
if unexplained != 0:
raise ValueError(f'IDENTITY: {sku}: effects do not match change')
values = [r0, r1, price, quantity, new, discontinued, change]
record = {
'sku': sku, 'classification': classification,
'previous_quantity': q0 if sku in previous else '',
'previous_price': amount(p0) if sku in previous else '',
'current_quantity': q1 if sku in current else '',
'current_price': amount(p1) if sku in current else '',
'currency': currency, 'unexplained': amount(unexplained),
}
for key, value in zip(AMOUNTS, values):
record[key] = amount(value)
totals[key] += value
rows.append(record)
unexplained = totals['change'] - sum(
(totals[key] for key in AMOUNTS[2:6]), ZERO)
if unexplained != 0:
raise ValueError('IDENTITY: aggregate effects do not match change')
summary = {'currency': currency,
**{key: amount(value) for key, value in totals.items()},
'unexplained': amount(unexplained)}
return rows, summary
def write_csv(path, fields, rows):
with path.open('w', encoding='utf-8-sig', newline='') as f:
writer = csv.DictWriter(f, fieldnames=fields, lineterminator='\n')
writer.writeheader()
writer.writerows(rows)
def main(argv=None):
parser = argparse.ArgumentParser(description='Two-period synthetic sales bridge')
parser.add_argument('--previous', type=Path, required=True)
parser.add_argument('--current', type=Path, required=True)
parser.add_argument('--currency', required=True)
parser.add_argument('--output-dir', type=Path, required=True)
args = parser.parse_args(argv)
stage = None
try:
if not re.fullmatch(r'[A-Z]{3}', args.currency):
raise ValueError('CURRENCY: supply one 3-letter uppercase code')
out = args.output_dir
if out.exists():
raise ValueError(f'OUTPUT_EXISTS: choose a new directory: {out}')
if not out.parent.is_dir():
raise ValueError(f'OUTPUT_PARENT: parent directory does not exist: {out.parent}')
paths = [args.previous, args.current]
snapshots = [read_snapshot(p) for p in paths]
hashes = [hashlib.sha256(raw).hexdigest() for raw in snapshots]
previous = read_period(args.previous, args.currency, snapshots[0])
current = read_period(args.current, args.currency, snapshots[1])
rows, totals = calculate(previous, current, args.currency)
if hashes != [hashlib.sha256(read_snapshot(p)).hexdigest() for p in paths]:
raise ValueError('INPUT_CHANGED: inputs changed during this run')
policy = {
'synthetic_practice': True,
'python_version': sys.version.split()[0],
'currency': args.currency,
'price_first': 'q0 * (p1 - p0)',
'quantity_second': 'p1 * (q1 - q0)',
'new_discontinued_basis': 'key absence, not quantity zero',
'rounding': 'none; input price maximum 2 decimal places',
'decimal_precision': 50,
'maximum_input_bytes_each': MAX_BYTES,
'status_policy': 'posted only; cancellation/returns rejected',
'previous_records': len(previous),
'current_records': len(current),
'result_records': len(rows),
'input_sha256': dict(zip(['previous', 'current'], hashes)),
}
stage = Path(tempfile.mkdtemp(prefix='.sales-bridge-', dir=out.parent))
write_csv(stage / 'bridge_rows.csv', ROW_FIELDS, rows)
write_csv(stage / 'bridge_totals.csv', TOTAL_FIELDS, [totals])
(stage / 'run_policy.json').write_text(
json.dumps(policy, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
if out.exists():
raise ValueError(f'OUTPUT_EXISTS: choose a new directory: {out}')
stage.rename(out)
stage = None
print(f"OK: previous={totals['previous_revenue']} "
f"current={totals['current_revenue']} change={totals['change']}")
print(f"effects={totals['price_effect']},{totals['quantity_effect']},"
f"{totals['new_effect']},{totals['discontinued_effect']} unexplained={totals['unexplained']}")
return 0
except (ValueError, OSError, UnicodeError, csv.Error) as error:
print(f'ERROR: {error}', file=sys.stderr)
return 2
finally:
if stage is not None:
shutil.rmtree(stage)
if __name__ == '__main__':
raise SystemExit(main())
재현 확인과 적용 한계
실습 폴더에서 다음 명령을 실행합니다.
python -m unittest -v
제작 검증에서는 23개 테스트가 통과했습니다. 한 테스트 안에서 고정 난수 시드의 500개 합성 사례를 별도의 정수 소단위 계산과 대조했습니다. 공통·신규·중단·관측 수량 0, 소수 단가, 중복 키, 잘못된 수량·단가·상태·통화·열 구조, 입력 행·바이트 상한, CLI 종료코드, 기존 결과 보존과 입력 SHA 보존을 확인했습니다. 검수한 ZIP을 새 폴더에 풀어 정상·오류 명령과 테스트를 다시 실행한 기록도 남깁니다. 난수 검산이 현실 자료의 모든 조건을 보장하지는 않습니다.
기술 문서는 2026년 10월 13일 한국시간에 확인했습니다. 온라인 Python 3.12 문서에는 3.12.15가 표시됐으며 실제 코드 실행은 CPython 3.12.14입니다. argparse 공식 문서는 명령행 인수 처리를 설명합니다. Excel 앱, 외부 서비스, 실제 매출 시스템에서는 실행하지 않았습니다.
완료 기준은 입력에 승인된 같은 SKU·단위·통화·단가 기준을 쓰고, 각 SKU와 전체의 네 효과 합이 매출 증감에 맞는지 확인하는 것입니다. 세금·외화 환산·수익 인식·반품 상계·상품 믹스의 별도 효과·기간 길이 조정·자동 보고·성과평가는 이 예제에 포함하지 않습니다. 가격과 수량이 함께 변한 이유를 알아내려면 판매 조건과 자료 수집 범위 등 별도의 근거가 필요합니다.
자주 묻는 질문
Excel 파일을 직접 실행하나요?
입력·출력은 CSV이고 계산은 Python에서 합니다. Excel 앱 실행이나 셀 수식 계산은 검증하지 않았습니다. CSV의 값을 Excel에서 표시하는 방식과 이 계산 결과의 산술 일치는 구분해 확인하세요.
전기 수량이 0이면 무조건 신규인가요?
두 파일에 SKU가 모두 있으면 공통 품목입니다. 신규·중단은 키 부재로 정합니다. 이 예제는 0기저 증감률을 구하지 않고 금액 차이를 연결합니다.
효과 합이 맞으면 실제 매출 원인도 설명됐나요?
합계 일치는 고정한 분해 순서의 산술 검산입니다. 가격 변경과 수량 변화의 실제 인과관계, SKU의 출시·중단 여부 또는 평가 기준의 타당성을 증명하지 않습니다.
함께 확인할 자료
- CSV 행별 금액이 수량×단가와 맞는지 Python으로 검산하기: 원본의 한 행 금액부터 따로 확인할 때
- 소수 금액 CSV를 Decimal로 계산하고 반올림 기준 남기기: 소수 표기와 반올림 정책을 검토할 때
- 엑셀 피벗테이블로 월별 매출 요약표를 만들고 검산하기: 월별 매출 합계와 거래 범위를 확인할 때
작성자 소개
