소수 금액 CSV를 Decimal로 계산하고 반올림 기준 남기기

CSV의 0.10을 바로 float로 바꾸기 전에, 어떤 통화와 금액 표기를 받을지 먼저 정하세요. 문자열을 Decimal로 직접 읽으면 입력한 십진 금액으로 합산할 수 있고, 마지막에 반올림했는지 행마다 반올림했는지도 따로 기록할 수 있습니다.

이 실습의 네 금액 0.10, 0.20, 1.005, 1.005는 원합계가 2.310입니다. 합산 후 소수 둘째 자리로 반올림하면 2.31, 행마다 먼저 반올림하면 합계가 2.32로 달라집니다. 원본 CSV는 그대로 두고 이 차이와 적용 규칙을 새 JSON 보고서에 남깁니다.

금액과 USD 표시는 가상 연습 입력입니다. 아래 통화·범위·반올림 규칙은 이 CSV 실습의 계약이며 실제 청구·세금·회계 규정을 뜻하지 않습니다.

1. 숫자로 바꾸기 전에 입력 계약 확인하기

ZIP을 새 빈 폴더에 풀고 input/amounts.csv를 텍스트 편집기로 엽니다. Excel로 열고 다시 저장하면 ID나 표기가 달라질 수 있으므로 첫 실습에서는 제공 파일을 그대로 사용하세요.

record_id,currency,amount
0001,USD,0.10
0002,USD,0.20
0003,USD,1.005
0004,USD,1.005

record_id는 계산할 숫자가 아닙니다. 0001을 정수 1로 바꾸지 않고 원문과 행 순서를 보존합니다. currency도 금액에 붙여 쓰지 않고 별도 열로 검사합니다.

• 파일: UTF-8 또는 UTF-8 BOM, 쉼표 구분 CSV, 2,000,000바이트 이하
• 구조: 머리글이 정확히 record_id,currency,amount 순서이며 모든 데이터 레코드는 세 열
• ID: ASCII 영문·숫자·밑줄·하이픈만 1~32자, 대소문자를 구분하며 중복 거부
• 통화: 정확히 USD만 허용, 환율 변환이나 다른 통화와의 합산 없음
• 금액: 부호 없는 ASCII 정수 또는 일반 소수, 정수부의 불필요한 앞자리 0 없이 소수부는 1~4자리
• 금액 범위: 0 이상 999999.9999 이하, 데이터는 1~10,000행
• 처리: 공백 제거·기호 제거·잘못된 값의 0 대체 없이, 한 행이라도 조건에 맞지 않으면 전체 중단

0, 0.00, 1.005는 받지만 +1.00, -0, 01.00, .50, 1., 1e2, 1_000, NaN, Infinity, 빈칸은 받지 않습니다. 1,000은 따옴표로 한 필드에 넣어도 허용 금액 표기가 아닙니다. USD, KRW, usd를 같은 단위로 취급하지도 않습니다.

Python 3.12 csv 공식 문서에 따르면 기본 reader는 문자열 필드를 반환합니다. 이 예제는 인용하지 않은 필드를 float로 바꾸는 QUOTE_NONNUMERIC 옵션을 사용하지 않습니다. strict=True로 CSV 구문 오류를 확인한 뒤, 열 수와 위 계약을 별도로 검사합니다.

2. Decimal 정밀도와 반올림 자리를 나누기

Decimal(amount_text)에 CSV에서 읽은 문자열을 바로 넘깁니다. Decimal(float(amount_text))처럼 float를 거치면 이미 이진 부동소수점으로 바뀐 값을 받게 됩니다. Decimal 생성자 공식 문서에서 두 입력 방식의 차이를 확인할 수 있습니다.

Decimal 생성자 자체는 지수 표기와 특수값도 받을 수 있습니다. 그래서 “Decimal 변환이 됐으니 정상 금액”으로 판단하지 않고, 앞의 표기 검사와 is_finite()·범위 확인을 먼저 통과시킵니다.

prec=28은 계산의 유효숫자 정밀도입니다. 소수 28자리로 보고서를 만들겠다는 뜻이 아닙니다. 이 계약의 원합계 상한은 999999.9999 × 10000 = 9999999999.0000이며 정수부 10자리와 소수부 4자리, 최대 14자리면 합산을 표현할 수 있습니다. 행별 반올림 후 합계 상한 10000000000.00도 13자리라 28자리 안에 들어갑니다. 음수를 받지 않아 중간 합계도 이 상한을 넘지 않습니다.

합산 중에는 Inexact 예외를 켜고, 의도적으로 반올림하는 단계에서는 끕니다. 반올림은 quantize(Decimal('0.01'), rounding=ROUND_HALF_UP)로 별도 지정합니다. quantize와 반올림 모드에 관한 공식 설명처럼 목표 지수와 모드를 명시하는 구조입니다.

여기서는 가장 가까운 값으로 반올림하고 정확한 중간값이면 0에서 멀어지는 ROUND_HALF_UP을 선택합니다. 따라서 양수 1.005를 소수 둘째 자리로 처리하면 1.01입니다. ROUND_HALF_UP 공식 정의는 이 선택을 설명하며 모든 통화에 이 모드를 적용하라는 규정은 아닙니다.

3. 문자열 금액을 검사하고 합산하는 전체 코드

다음 코드는 ZIP의 code/decimal_csv.py와 동일합니다. Python 3.10 이상 표준 라이브러리만 사용하며 별도 패키지 설치는 없습니다. 실제 검수 환경은 Linux의 Python 3.12.14입니다.

"""Sum fictional monetary CSV strings with an explicit rounding stage."""
import argparse
import csv
import hashlib
import io
import json
import re
import sys
from decimal import Context, Decimal, DecimalException, FloatOperation, Inexact, ROUND_HALF_UP, localcontext
from pathlib import Path

HEADER = ['record_id', 'currency', 'amount']
ID_PATTERN = re.compile(r'[A-Za-z0-9_-]{1,32}\Z')
AMOUNT_PATTERN = re.compile(r'(?:0|[1-9][0-9]{0,6})(?:\.[0-9]{1,4})?\Z')
MAX_AMOUNT = Decimal('999999.9999')
QUANTUM = Decimal('0.01')
MAX_ROWS = 10_000
MAX_BYTES = 2_000_000


def inspect(source):
    with Path(source).open('rb') as handle:
        raw_bytes = handle.read(MAX_BYTES + 1)
    if len(raw_bytes) > MAX_BYTES:
        raise ValueError('바이트 한도 2,000,000을 초과했습니다')
    try:
        text = raw_bytes.decode('utf-8-sig', errors='strict')
    except UnicodeDecodeError as error:
        raise ValueError('UTF-8 해석 실패: 인코딩을 확인하세요') from error

    rows, seen = [], set()
    try:
        reader = csv.reader(io.StringIO(text, newline=''), strict=True)
        if next(reader, None) != HEADER:
            raise ValueError('머리글은 record_id,currency,amount 순서여야 합니다')
        for record_number, fields in enumerate(reader, 2):
            if len(fields) != 3:
                raise ValueError(f'레코드 {record_number}: 열 수는 3개여야 합니다')
            ident, currency, amount_text = fields
            if not ID_PATTERN.fullmatch(ident):
                raise ValueError(f'레코드 {record_number}: ID 형식 확인')
            if ident in seen:
                raise ValueError(f'레코드 {record_number}: 중복 ID {ident}')
            seen.add(ident)
            if currency != 'USD':
                raise ValueError(f'레코드 {record_number}: 허용 통화는 USD입니다')
            if not AMOUNT_PATTERN.fullmatch(amount_text):
                raise ValueError(f'레코드 {record_number}: 금액 표기 확인')
            amount = Decimal(amount_text)
            if not amount.is_finite() or not Decimal('0') <= amount <= MAX_AMOUNT:
                raise ValueError(f'레코드 {record_number}: 금액 범위 확인')
            rows.append({'record_id': ident, 'currency': currency,
                         'raw_amount': amount_text, 'amount': amount,
                         'source_record': record_number})
            if len(rows) > MAX_ROWS:
                raise ValueError('행 한도 10,000을 초과했습니다')
    except csv.Error as error:
        raise ValueError('CSV 구문 오류: ' + str(error)) from error
    if not rows:
        raise ValueError('데이터 행이 1개 이상 필요합니다')

    try:
        # At most 10,000 rows x 999999.9999: raw sum needs at most 14 digits.
        with localcontext(Context(prec=28)) as context:
            context.traps[FloatOperation] = True
            context.traps[Inexact] = True
            raw_total = sum((row['amount'] for row in rows), Decimal('0'))
            # Rounding is intentional only at the next two documented stages.
            context.traps[Inexact] = False
            rounded_total = raw_total.quantize(QUANTUM, rounding=ROUND_HALF_UP)
            rounded_rows = [row['amount'].quantize(QUANTUM, rounding=ROUND_HALF_UP)
                            for row in rows]
            row_total = sum(rounded_rows, Decimal('0.00'))
            delta = row_total - rounded_total
    except DecimalException as error:
        raise ValueError('Decimal 계산 실패: 범위와 정밀도 정책을 확인하세요') from error

    output_rows = []
    for row, rounded in zip(rows, rounded_rows):
        output_rows.append({key: value for key, value in row.items() if key != 'amount'})
        output_rows[-1]['rounded_amount'] = format(rounded, 'f')
    return {
        'currency': 'USD', 'row_count': len(rows),
        'raw_total': format(raw_total, 'f'),
        'total_after_rounding': format(rounded_total, 'f'),
        'sum_after_row_rounding': format(row_total, 'f'),
        'row_rounding_delta': format(delta, 'f'),
        'policy': {
            'quantum': '0.01', 'rounding': 'ROUND_HALF_UP',
            'primary_rounding_stage': 'after_total',
            'comparison_rounding_stage': 'each_row_then_sum',
            'delta_definition': 'row_total_minus_rounded_total',
            'context_precision': 28, 'input_fractional_digits_max': 4,
            'amount_min': '0', 'amount_max': '999999.9999',
            'signs_allowed': False, 'max_rows': MAX_ROWS,
        },
        'rows': output_rows,
        'source_sha256': hashlib.sha256(raw_bytes).hexdigest(),
        'source_changes_written': 0,
    }


def write_report(source, target):
    source, target = Path(source), Path(target)
    if source.resolve() == target.resolve():
        raise ValueError('입력과 보고서는 서로 다른 경로여야 합니다')
    if target.suffix.lower() != '.json':
        raise ValueError('보고서는 새 .json 파일로 지정하세요')
    if target.exists() or target.is_symlink():
        raise ValueError('기존 보고서 보존: 새 --report 이름을 지정하세요')
    report = inspect(source)
    rendered = json.dumps(report, ensure_ascii=False, indent=2) + '\n'
    target.parent.mkdir(parents=True, exist_ok=True)
    created = False
    try:
        with target.open('x', encoding='utf-8', newline='\n') as handle:
            created = True
            handle.write(rendered)
    except FileExistsError as error:
        raise ValueError('기존 보고서 보존: 새 --report 이름을 지정하세요') from error
    except OSError:
        if created:
            target.unlink(missing_ok=True)
        raise
    return report


def main():
    parser = argparse.ArgumentParser(description='가상 금액 CSV의 Decimal 합계·반올림 시점 검산')
    parser.add_argument('--input', required=True)
    parser.add_argument('--report', required=True)
    args = parser.parse_args()
    try:
        report = write_report(args.input, args.report)
    except (ValueError, OSError) as error:
        print('오류: ' + str(error), file=sys.stderr)
        return 2
    print(f"원합계 {report['raw_total']} / 합계 후 {report['total_after_rounding']} / "
          f"행별 후 합계 {report['sum_after_row_rounding']} / 차이 {report['row_rounding_delta']}")
    return 0


if __name__ == '__main__':
    raise SystemExit(main())

inspect()는 읽기·검사·계산, write_report()는 새 보고서 저장을 담당합니다. 보고서의 금액은 JSON 숫자 대신 문자열로 저장해 2.310과 소수 둘째 자리 결과를 구분합니다. 다시 계산할 때도 이 문자열을 Decimal로 읽는 계약이 필요합니다.

source_record는 머리글을 1로 센 논리 레코드 번호입니다. 따옴표 안 줄바꿈이 있는 CSV에서는 화면의 물리 줄 번호와 다를 수 있습니다. ID와 raw_amount는 변환 전 문자열, rounded_amount는 행별 비교용 결과입니다.

4. 첫 실행에서 2.310과 2.31과 2.32 확인하기

터미널에서 ZIP을 푼 폴더로 이동해 실행합니다. 명령의 python이 Python 3.10 이상을 가리키는지 첫 줄로 확인하세요. 명령을 찾지 못하면 설치된 Python의 명령 이름을 확인한 뒤 같은 이름을 아래 명령에도 사용합니다.

python --version
python code/decimal_csv.py --input input/amounts.csv --report out/amounts.json

out 폴더는 필요하면 자동으로 만듭니다. 정상 실행의 종료코드는 0이며 다음 문구가 나옵니다.

원합계 2.310 / 합계 후 2.31 / 행별 후 합계 2.32 / 차이 0.01

out/amounts.json을 열고 아래 필드와 행별 값을 대조합니다.

행별 결과의 합은 0.10 + 0.20 + 1.01 + 1.01 = 2.32입니다. 비교 결과를 원합계에 섞지 마세요. policy.primary_rounding_stage는 after_total, policy.rounding은 ROUND_HALF_UP, policy.quantum은 0.01이어야 합니다. 합계가 맞는지와 어떤 규칙으로 만들었는지를 같이 확인합니다.

2.310과 2.31은 수치로 같아도 보고서에서 맡은 역할이 다릅니다. 두 문자열만 보고 계산 정확성을 판단하지 말고 원문·반올림 단계·기대값을 함께 읽으세요.

여러 파일을 모을 때는 반올림 범위도 정합니다

이 코드의 after_total은 입력 CSV 한 개의 합계를 반올림한다는 뜻입니다. 여러 파일의 total_after_rounding을 더하면 파일마다 먼저 반올림한 합계가 됩니다. 월 전체의 원금액을 모두 더한 뒤 한 번 반올림하기로 한 규칙과 결과가 달라질 수 있습니다.

이 글의 ROUND_HALF_UP·소수 둘째 자리 규칙으로, 같은 네 금액을 계산 범위만 나눠 보겠습니다. A묶음 0.10 + 1.005 = 1.105는 반올림 후 1.11, B묶음 0.20 + 1.005 = 1.205는 반올림 후 1.21입니다. 두 결과를 더하면 2.32지만 네 원금액을 먼저 더한 2.310을 한 번 반올림하면 2.31입니다. 이는 같은 금액을 나눈 산술 예시이며 ZIP에 별도의 A·B 입력 파일이 있다는 뜻은 아닙니다. quantize 공식 설명은 지정한 자리로 반올림하는 동작을 설명하고, 어느 범위에서 적용할지는 업무 규칙에서 정해야 합니다.

월말 금액 확인에는 합계 계산 전에 받아야 할 파일과 확정된 대상 목록이 필요합니다. 날짜별 수신이 기준이라면 날짜별 파일 수신 대조의 적용 범위부터 확인하세요. 이 Decimal 코드는 한 파일 안의 중복 ID를 거부하지만 파일 간 ID 중복이나 빠진 파일은 찾지 않습니다. 통화·허용 표기·대상 파일·ID의 고유 범위·반올림 단위와 시점이 미정이면 최종 월 합계를 확정하지 않습니다. 여러 파일을 합치는 프로그램은 이 ZIP에 포함되지 않으며, 검증 없이 파일별 보고서의 반올림 결과를 더해 월 전체 결과로 사용하지 마세요.

5. 0과 범위 끝에서도 같은 계약 적용하기

다음 세 명령은 서로 다른 새 보고서를 만듭니다.

python code/decimal_csv.py --input input/zero.csv --report out/zero.json
python code/decimal_csv.py --input input/boundary.csv --report out/boundary.json
python code/decimal_csv.py --input input/precision4.csv --report out/precision4.json

범위 검사는 반올림 전 입력 금액에 적용합니다. 그래서 입력 1000000은 거부하지만 입력 999999.9999에서 나온 결과 1000000.00은 정상입니다. 출력 상한도 별도로 제한해야 하는 업무라면 그 조건을 먼저 확정해야 합니다.

6. 오류가 나면 표기·통화·구조부터 복구하기

아래 명령은 각각 종료코드 2로 멈추고 지정한 새 보고서가 없어야 합니다.

python code/decimal_csv.py --input input/failures/nan.csv --report out/rejected_nan.json
python code/decimal_csv.py --input input/failures/unapproved_currency.csv --report out/rejected_currency.json
python code/decimal_csv.py --input input/failures/precision5.csv --report out/rejected_precision.json
python code/decimal_csv.py --input input/failures/duplicate_id.csv --report out/rejected_id.json

• 금액 표기 오류: 해당 레코드의 원문을 확인합니다. NaN을 0으로 바꾸거나 공백을 조용히 제거하지 않습니다
• 통화 오류: KRW나 USD/kg가 섞였는지 확인합니다. 값을 그대로 두고 통화 이름만 USD로 바꾸면 환산이 된 것이 아닙니다
• 소수 5자리 오류: 원본을 임의로 잘라 맞추지 말고 허용 입력 정밀도부터 확인합니다
• 중복 ID 오류: 중복이 왜 생겼는지 확인하기 전에 두 행을 합치거나 하나를 지우지 않습니다
• UTF-8·CSV 구문·열 수 오류: 공급 파일의 인코딩, 닫히지 않은 따옴표, 빠지거나 추가된 열을 사본에서 확인합니다

오류 확인은 작업용 사본에서 진행하고 정상 가상 파일로 다시 연습하세요. 이 코드는 일부 정상 행의 소계만 저장하지 않습니다. 잘못된 행이 있으면 전체 합산 보고서를 보류합니다.

앞에서 만든 out/amounts.json에 같은 명령을 다시 실행하면 기존 보고서를 보존하고 멈춥니다. 새 결과가 필요하면 이름을 바꿉니다.

python code/decimal_csv.py --input input/amounts.csv --report out/amounts_again.json

Python 파일 열기 공식 문서의 배타적 생성 모드 x를 사용합니다. 기존 파일·링크와 입력과 같은 경로는 출력 대상으로 받지 않습니다. 입력 SHA-256은 읽은 원본 바이트의 기록이며 원본을 잠그는 기능은 아닙니다.

프로그램이 잡아낸 쓰기 오류에서는 이번 실행이 만든 새 파일을 제거하려고 합니다. 검수에서는 일부 내용을 쓴 뒤 OSError를 발생시키는 시험으로 새 출력 부재를 확인했습니다. 강제 종료·정전·동시 실행·파일 제거 실패까지 보호하는 구조는 아니므로 오류나 중단 뒤 남은 파일을 완료 보고서로 사용하지 마세요.

7. 실습 ZIP과 직접 재검산하기

실습 파일: P07_Practice_Decimal_CSV.zip

검수된 Decimal CSV 실습 ZIP 다운로드

ZIP에는 전체 코드, 정상 CSV 5개, 실패 CSV 17개, 고정 기대 JSON, README, 시험 코드, 실제 실행 로그와 SHA-256 목록이 들어 있습니다. expected/amounts.json은 네 금액의 전체 기대 보고서, expected/summaries.json은 정상 다섯 파일의 합계 기대값입니다. 입력을 바꾼 다음 이 고정 정답까지 같이 고치면 원래 실습의 검산이 되지 않습니다.

python -m unittest discover -s code -p "test_decimal_csv.py" -v
python code/verify_examples.py

첫 명령은 unittest의 discover로 시험 32개를 실행합니다. 정상·0·최대금액·10,000행·표기/단위 거부·CSV 오류·기존 출력·재실행·원본 보존과 처리된 쓰기 오류를 확인합니다. 일부 시험은 여러 반례를 반복하는 하위시험을 포함합니다.

두 번째 명령은 매번 새 임시 폴더에 가상 입력을 복사해 CLI 25건을 실행하고, 고정 기대값·종료코드·실패 출력 부재·기존 출력 보존·원본 바이트 보존을 대조합니다. 실제 검수에서 두 검사 모두 통과했고 입력 CSV 22개의 실행 전후 SHA-256이 같았습니다. Windows 실행과 Python 3.10·3.11 실행은 확인하지 않았습니다.

업무 CSV로 바꾸기 전에는 통화와 허용 표기, 환불 등 부호의 의미, 입력 정밀도, 반올림 모드와 적용 시점을 먼저 확정하세요. 이 실습은 그 규칙이 정해진 파일을 정확하게 읽고 계산 근거를 남기는 연습입니다.

공식 자료 확인일: 2026-10-05. 인용 링크는 Python 3.12 계열 문서이며 실제 실행 버전 3.12.14와 구분합니다.

작성자 소개

최지원

엑셀·VBA·파이썬·AI 업무 자동화의 사용 절차와 실습 자료를 정리합니다.

WP 게시물 작성자의 글 보기