• 주제: AI 기반 개인 재무 관리 및 투자 포트폴리오 추천 플랫폼 WooriPort (우리포트)

• 프로젝트 기획 배경:

  모바일 금융서비스 이용 경험자는 95% 이상이지만, 20대 금융이해력은 62.6점(전 연령대 2위), 디지털 금융이해력은 44.8점에 그칩니다. (2024 금융감독원·한국은행)

  쓰는 법은 알지만 굴리는 법은 모르는 세대입니다.

  그러면서도 재무목표로 자산 증식을 꼽는 비율은 2022년 대비 3배 가까이 급증(7.3% → 19.9%)했고, 목표 달성 수단으로 저축·투자(84.1%) 와 지출 축소(47.4%) 를 꼽았습니다.

  목표는 뚜렷한데 방법을 모르는 2030을 위해 WooriPort를 기획했습니다. 마이데이터로 자산 현황을 자동으로 불러오고, LangGraph 기반 멀티 에이전트가 투자 성향과 소비 패턴을 분석해 "이 돈은 여기에 넣으세요"까지 제시합니다.

• 기술 스택

  • AI Server: LangGraph, OpenAI API (GPT-4o), PyPortfolioOpt (HRP), Tavily, yfinance
  • Backend: Spring Boot 3.5.x, Spring Batch, Spring Security, Spring Data JPA, FastAPI, Apache Kafka, Redis, JWT 인증
  • Frontend: React 19, TypeScript, Vite, Tailwind CSS, Custom SVG Chart
  • Database: PostgreSQL (pgvector 확장), Redis
  • Data Pipeline: Apache Kafka, Spring Batch Scheduler
  • Infra: AWS EC2, Docker, Docker Compose, Terraform, ELK Stack, Prometheus & Grafana
  • CI/CD: AWS CLI, Local Build & Sync (S3/CloudFront)

Frontend(React)와 Backend(Spring Boot), AI Server(FastAPI)가 각각 독립적인 컨테이너로 구성되며, PostgreSQL, Redis, Kafka와 연동됩니다. 사용자 요청은 Backend를 통해 인증 및 데이터 처리가 이루어지고, AI 추천이 필요한 경우 AI Server로 위임됩니다. AI Server는 LangGraph PEV(Planner → Executor → Verifier) 패턴으로 투자 흐름별 에이전트를 병렬 실행하며, 결과를 Backend를 통해 DB에 저장합니다. Kafka 거래 이벤트는 Logstash → Elasticsearch → Kibana 파이프라인으로 수집·시각화되고, Prometheus/Grafana를 통한 메트릭 수집으로 시스템 운영을 모니터링합니다. 운영 환경은 Terraform으로 AWS 인프라(EC2, RDS, ALB, CloudFront)를 프로비저닝하고, Jenkins 파이프라인으로 배포합니다.

자산 포트폴리오 추천은 PEV(Planner → Executor → Verifier) 패턴으로 동작합니다.

• Planner: 투자 성향·관심사·보유 계좌를 종합해 단기/중기/장기 흐름(flow_plans)을 설계하고 계좌를 선택합니다.
• Executor: 각 흐름을 독립 서브그래프로 병렬 실행합니다. LangChain create_agent로 ETF를 pgvector RAG 검색 후 선택합니다.
• Verifier: HRP 알고리즘으로 ETF 비중을 최적화하고, 복리 계산으로 기대 수익을 산출해 최종 투자 흐름을 조립합니다.

LangGraph Send API로 흐름별 서브그래프를 Fan-out 병렬 실행 → Annotated[list, operator.add]로 결과를 Fan-in 집계합니다.

• 기능 설명: 사용자의 PorTI 투자 성향, 인생 목표(내집마련·노후 등), 보유 계좌를 종합 분석해 단기·중기·장기 투자 흐름을 설계하고 각 흐름에 최적화된 ETF 포트폴리오를 추천합니다. LangGraph PEV(Planner–Executor–Verifier) 패턴으로 흐름별 병렬 실행하며, HRP 비중 최적화로 기대 수익을 산출합니다.
• 핵심 코드:

# app/services/agent/asset_portfolio.py:645-663
def _build_graph() -> StateGraph:
    flow_sg = StateGraph(FlowState)
    flow_sg.add_node("executor", _node_executor)
    flow_sg.add_node("verifier", _node_verifier)
    flow_sg.set_entry_point("executor")
    flow_sg.add_edge("executor", "verifier")
    flow_sg.add_edge("verifier", END)

    graph = StateGraph(AssetPortfolioState)
    graph.add_node("planner", _node_planner)
    graph.add_node("flow_branch", flow_sg.compile())
    graph.set_entry_point("planner")
    graph.add_conditional_edges("planner", _route_flows, ["flow_branch"])
    graph.add_edge("flow_branch", END)

    return graph.compile()

• 코드 링크: ai-server/app/services/agent/asset_portfolio.py

• 기능 설명: 월급에서 고정지출을 제외한 가처분소득을 사용자의 소비 패턴과 보유 계좌 잔액에 맞게 생활비·비상금·예비비·투자금으로 재배분합니다. LangGraph Plan-Reflect 패턴(최대 2회 재계획)으로 분산 배분 실패·투자금 허용 범위 이탈·용도-계좌 불일치 등을 자동 심사하고, PorTI 성향별 투자금 허용 범위를 코드로 강제해 LLM의 과도한 투자 배분을 방지합니다.
• 핵심 코드:

# app/services/agent/rebalance.py:492-499
def _build_graph() -> StateGraph:
    graph = StateGraph(RebalanceState)
    graph.add_node("plan", _plan_rebalance)
    graph.add_node("reflect", _reflect)
    graph.set_entry_point("plan")
    graph.add_edge("plan", "reflect")
    graph.add_conditional_edges(
        "reflect", _route_after_reflect, {"plan": "plan", END: END}
    )
    return graph.compile()

# app/services/agent/rebalance.py:22-26
_INVEST_RATIO: dict[str, tuple[float, float]] = {
    **{t: (0.10, 0.25) for t in STABLE_PORTI_TYPES},
    **{t: (0.15, 0.35) for t in NEUTRAL_PORTI_TYPES},
    **{t: (0.25, 0.50) for t in INVEST_PORTI_TYPES},
}

• 코드 링크: ai-server/app/services/agent/rebalance.py

• 기능 설명: 이번 달 소비 내역을 카테고리별로 분석해 가장 절약 가능성이 높은 항목의 미니 챌린지를 자동 제안하고, 절약한 금액을 주식 투자 리워드로 연결합니다. Redis 세션(7일 TTL)에 제안 이력을 누적 저장해 동일 챌린지 반복 제안을 방지하며, 사용자 피드백("더 쉽게/어렵게/주제 변경")에 따라 챌린지를 실시간으로 조정합니다.
• 핵심 코드:

# app/services/agent/mini_challenge_agent.py:116-156
async def propose_mini_challenge(req: MiniChallengeRequest) -> MiniChallengeResponse:
    session = await get_session(req.user_id)

    cats = "\n".join(
        f"- {c.category}({c.sender_name}): {c.amount:,}원"
        for c in sorted(req.category_expense, key=lambda x: x.amount, reverse=True)
    )
    prices_text = await _fetch_prices_text()

    context = (
        f"관심 주식 테마: {themes_text}\n\n"
        f"이번 달 소비 패턴:\n{cats}\n\n"
        f"현재 주가:\n{prices_text}"
    )

    result = await ainvoke_structured(
        [SystemMessage(content=_INIT_SYSTEM), HumanMessage(content=context)],
        _MiniChallengeAIOutput,
    )

    if result:
        session.setdefault("proposals", []).append({**result.model_dump(), "feedback": ""})
        await save_session(req.user_id, session)
        return MiniChallengeResponse(
            created_at=datetime.now(timezone.utc),
            ticker_name=ticker_to_name(data.get("ticker")) or "",
            **data,
        )

• 코드 링크: ai_server/app/services/agent/mini_challenge_agent.py

• 기능 설명: transaction-events 컨슈머를 4단계로 개선해 처리량과 정합성을 동시에 확보했습니다.

  단계	핵심 변경	처리량	주요 성과
  Phase 0 (Baseline)	단건 처리 / 단일 스레드	~301 msg/s	병목 및 정합성 약점 측정
  Phase 1 (Resilience)	event_id 멱등 적재 + 급여 재처리 토픽(DLT) + 챌린지 DB 재계산	~248 msg/s	중복 행 0, 급여 유실 0, 챌린지 오차 자가복구 — 통합테스트 4개 green
  Phase 2 (Batch)	@KafkaListener(batch=true) + 2-레이어 dedup + saveAll 배치 INSERT + poll 단위 단일 TX	회복 목표	DB 라운드트립 2,000 ops → 4 ops (poll 500건 기준)
  Phase 3 (Tuning)	max-poll-records=1000 + partitions=3 + concurrency=3 스윕	~1,486 msg/s	Baseline 대비 약 5배, concurrency=6 은 DB 경합으로 역전 확인

  Phase 1의 -18% 처리량 하락(멱등 적재 도입 비용)을 Phase 2 배치 insert로 회복하고, Phase 3 파라미터 스윕으로 최적값(mpr=1000, concurrency=3)을 수치로 확정했습니다. concurrency=6 실험에서 PostgreSQL 배치 INSERT 경합으로 persist p50이 0.19s → 1.14s로 6배 폭증해 역전되는 현상을 검증했습니다.

• 핵심 코드 (Phase 2 배치 컨슈머):

// Phase 2: poll 단위 배치 수신 + 2-레이어 dedup + saveAll
@KafkaListener(topics = "transaction-events",
               groupId = "approval-detect-group",
               containerFactory = "batchKafkaListenerContainerFactory")
public void consume(List<String> messages) {
    // 1. intra-batch dedup (동일 poll 내 중복 차단)
    // 2. DB dedup — findExistingEventIds(IN) 1쿼리
    // 3. findByAssetNumberIn(IN) 1쿼리
    // 4. saveAll — Hibernate jdbc.batch_size=100 + reWriteBatchedInserts
    BatchPersistResult result = transactionService.batchPersist(messages);
    // 후속처리(챌린지·급여)는 적재 커밋 후 per-item best-effort
}

• 코드 링크: backend/docs/kafka-exp/

• 기능 설명: 운영 관측을 로그와 메트릭 두 채널로 분리합니다.
  • ELK (비즈니스 로그): Backend가 Kafka service-logs 토픽으로 구조화 로그를 발행하면, Logstash가 actuator 경로를 필터링하고 amount 필드를 숫자로 변환해 Elasticsearch에 일별 인덱스(service-logs-YYYY.MM.dd)로 적재합니다. Kibana에서 거래 로그를 검색·집계합니다.
  • Prometheus + Grafana (시스템 메트릭): Prometheus가 Spring Boot(/actuator/prometheus)와 FastAPI(/metrics)에서 15초 주기로 메트릭을 수집합니다. Grafana 대시보드에서 총 요청 수(RPS), 에러율(4xx·5xx), 평균 응답 시간, URI별 지연, FastAPI P95 응답 시간, 이체 성공률을 실시간으로 시각화합니다.
• 핵심 설정:

# infra/logstash/pipeline/logstash.conf
input {
  kafka {
    bootstrap_servers => "kafka:29092"
    topics => ["service-logs"]
    codec => "json"
  }
}
filter {
  if [message] =~ "/actuator/" { drop {} }
  mutate { convert => { "amount" => "integer" } }
}
output {
  elasticsearch {
    hosts => ["http://elasticsearch:9200"]
    index => "service-logs-%{+YYYY.MM.dd}"
  }
}

# infra/prometheus/prometheus.yml
scrape_configs:
  - job_name: 'wooriport'          # Spring Boot
    metrics_path: '/actuator/prometheus'
    static_configs:
      - targets: ['host.docker.internal:8080']
  - job_name: 'fastapi'            # FastAPI
    metrics_path: '/metrics'
    static_configs:
      - targets: ['host.docker.internal:8000']

• 코드 링크: infra/logstash/pipeline/logstash.conf, infra/prometheus/prometheus.yml, infra/grafana/provisioning/dashboards/wooriport.json

fisache/
├── ai-server/           # FastAPI + LangGraph 에이전트 서버
│   └── app/
│       ├── routers/     # portfolio, salary, mini_challenge, report, consultant
│       ├── services/
│       │   ├── agent/   # PEV 에이전트 로직 (asset_portfolio, salary_rebalance, ...)
│       │   ├── rag/     # pgvector 기반 RAG (ETF 상품 검색)
│       │   └── ml/      # ML 모델 로더
│       ├── schemas/     # Pydantic 요청·응답 스키마
│       ├── kafka/       # Kafka 이벤트 발행
│       └── db/          # DB 커넥션 풀
├── backend/             # Spring Boot REST API 서버
│   └── src/main/java/com/wooriport/core_api/
│       ├── controller/  # 17개 컨트롤러 (Auth, Asset, Portfolio, Challenge, ...)
│       ├── service/     # 비즈니스 로직
│       ├── domain/      # JPA 엔티티 (Users, Portfolios, MiniChallenges, ...)
│       ├── repository/  # Spring Data JPA 레포지토리
│       ├── base/batch/  # Spring Batch (월간 리포트, 자산 스냅샷, 급여 이체)
│       └── config/      # Security, Redis, WebClient, Swagger
├── frontend/            # React 19 + TypeScript SPA
│   └── src/
│       ├── screens/     # 페이지 컴포넌트 (Dashboard, AssetPortfolio, SalaryManagement, ...)
│       ├── components/  # 재사용 UI (dashboard 위젯, 포트폴리오 모달, ...)
│       ├── api/         # API 클라이언트 모듈 (client.ts, portfolioApi.ts, ...)
│       └── contexts/    # AuthContext (JWT 상태 관리)
├── infra/               # Docker Compose + Terraform + Grafana/Prometheus/ELK 설정
│   ├── docker-compose.yml
│   ├── terraform/       # AWS 인프라 IaC (EC2, RDS, ALB, CloudFront, Jenkins)
│   ├── grafana/         # 대시보드·알림 프로비저닝
│   ├── prometheus/
│   └── logstash/        # Kafka → Elasticsearch 파이프라인
└── mock-server/         # 마이데이터 연동 목업 서버

운영 환경은 Terraform으로 AWS 인프라를 프로비저닝합니다: EC2(AI Server, Backend, Kafka, Monitoring), RDS(PostgreSQL), ALB, CloudFront(CDN), Jenkins(CI/CD).

• 코드 링크: infra/docker-compose.yml, infra/terraform/