투자회수 내역의 XIRR 값들을 일괄 업데이트 해야 하는 이슈가 있었는데 간결하고 심플한 파이썬으로 작업을 하게 되었습니다.
이미 Spring 웹서버에 단건의 투자건에 대해 XIRR을 구하는 로직은 구현이 되어 있었고 포팅만 하면 되는 상황이었는데 생산성을 높이고자 파이썬으로 작업을 하게 됩니다
XIRR 일괄 업데이트는 FAST API 를 발행해서 스케줄러를 통해 새벽 시간대에 업데이트를 하도록 설정하고 XIRR 산식은 채찍피티를 갈궈서 빠르게 클래스 객체로 만들어 내게 됩니다.
생산성은 기가 막히는데 딜레마에 빠지기도 하지만 MSDN 찾아보고 구글링 하던 시절 생각하면 별반 차이는 없다고 생각이 드네요
CashFlow 는 현금흐름 정보를 가지는 객체입니다
class CashFlow:
def __init__(self, date_str: str, amount: float):
# "YYYY-MM-DD" 형식의 문자열을 datetime 객체로 변환
self.date = datetime.strptime(date_str, "%Y-%m-%d")
self.amount = amount
XIRRCalculator 은 현금흐름의 순현재가치 xnpv를 계산하고 이를 통해 XIRR 을 구하는 클래스입니다
class XIRRCalculator:
def __init__(self, cash_flows: List[CashFlow]):
self.cash_flows = cash_flows
def xnpv(self, rate: float) -> float:
# 시작 날짜를 기준으로 일수 계산
start_date = self.cash_flows[0].date
npv = 0.0
for cf in self.cash_flows:
days = (cf.date - start_date).days
# amount와 days의 유효성 검사
try:
amount = float(cf.amount)
if days < 0 or rate == -1:
print(f"Invalid calculation for days: {days} or rate: {rate}")
continue # 유효하지 않은 값은 건너뜁니다.
npv += amount / ((1 + rate) ** (days / 365.0))
except (TypeError, ValueError) as e:
print(f"Invalid data encountered in xnpv calculation: {e}")
return npv
def calculate_xirr(self) -> float:
# 현금흐름에 양수와 음수가 포함되어 있는지 확인
has_positive = any(float(cf.amount) > 0 for cf in self.cash_flows)
has_negative = any(float(cf.amount) < 0 for cf in self.cash_flows)
if not (has_positive and has_negative):
print("XIRR 계산 불가: 현금흐름에 양수와 음수가 모두 포함되어야 합니다.")
return float('nan')
# 여러 초기 추정치를 시도하여 NaN 문제 해결
for guess in [0.1, 0.05, -0.05, 0.2, -0.2]:
try:
xirr = newton(lambda r: self.xnpv(r), guess)
return round(xirr * 100, 2) # 백분율로 변환 후 소수점 두 자리까지 반올림
except RuntimeError as e:
print(f"Guess {guess} failed to converge: {e}")
except Exception as e:
print(f"Unexpected error with guess {guess}: {e}")
print("XIRR 계산 실패: 모든 초기 추정치가 수렴하지 않았습니다.")
return float('nan')
투자건이 많아서 먼저 get_investlist 함수를 통해 투자건의 키값 리스트를 받아온 뒤 이 키를 통해 캐쉬플로우 데이터를 다시 받아와 XIRR을 구하도록 구현해두었습니다.
invest_list = self.get_investlist()
# 키컬럼 : insucd
# 투자리스트별 현금흐름 리스트 생성
for investkey in invest_list:
# 투자건별 현금흐름 리스트 조회
try:
cash_flows = self.db_manager_ams.get_invest_cashflow(investkey)
cash_flows_list = [CashFlow(cash_flow['investdate'], cash_flow['investmoney']) for cash_flow in cash_flows]
# XIRR 계산
calculator = XIRRCalculator(cash_flows_list)
xirr_result = calculator.calculate_xirr()
# Insucd, XIRR 결과 출력
#print(invest, xirr_result)
# XIRR 업데이트
self.db_manager.update_invest_xirr(investkey, xirr_result)
#print('************ XIRR 계산 : End %s, %s ************', insucd, xirr_result)
except Exception as e:
print('************ XIRR error : %s ', investkey)
print('exception : ', e)
print("Traceback:", traceback.format_exc())
continue
일괄 업데이트 된 데이터 몇개와 엑셀에서 제공하는 XIRR 함수 결과값을 비교해보니 동일한 값이 나오는 것을 확인했고 나머지 데이터 모두 업데이트 시키고, 데일리 스케쥴링을 통해 매일 업데이트 되도록 설정해두며 작업을 마무리 했네요