Python 함수 연결 예제
Python 함수가 Method로 등록되고, 선택한 데이터를 받아 결과를 내기까지의 과정을 살펴봅니다.
함수 작성 → Method 등록 → 입력 선택 → 데이터 조회 → 결과 확인
상품별 매출 순위로 기본 흐름을 보고, OLS 회귀분석으로 여러 변수를 연결하는 방법을 알아봅니다. 코드 아래에는 준비된 예제 데이터로 실행한 결과가 있습니다.
예제 데이터 준비 코드
import sys
from pathlib import Path
import pandas as pd
# examples/ is outside the installed decision_layer package.
repo_root = next(
path
for path in (Path.cwd(), *Path.cwd().parents)
if (path / "examples/methods").is_dir()
)
sys.path.insert(0, str(repo_root))
from decision_layer.methods import Column, DataInput, Scope, python_method
from importlib import reload
from examples.methods.authoring import customer_analysis
# Read updated example code when rerunning this setup cell.
practice_session = reload(customer_analysis).practice_session1. 상품별 매출 순위
먼저 간단한 순위 계산으로 함수와 데이터를 연결하는 과정을 살펴봅니다.
계산 함수
value 열의 순위를 계산하고 group 열과 함께 반환합니다. 함수는 Cube나 Decision Layer를 몰라도 됩니다.
def rank_values(data, descending=True):
"""Rank observed values, retaining ties."""
return data[["group", "value"]].assign(
rank=data["value"].rank(method="min", ascending=not descending).astype(int)
)
frame = pd.DataFrame({"group": ["A", "B", "C"], "value": [30, 20, 20]})
standalone = rank_values(frame)
standalone실행 결과
group value rank
0 A 30 1
1 B 20 2
2 C 20 2Method로 등록하기
python_method(...)로 함수에 이름과 입력 조건을 붙입니다.
name,version: Method의 이름과 버전DataInput: 함수에 전달할 데이터Column: 데이터에 필요한 열.measure는 지표,dimension은 차원입니다.
여기서는 value 지표와 group 차원이 필요하다고만 정합니다. 실제 매출 지표와 상품 차원은 실행할 때 선택합니다.
practice_session()은 예제 데이터가 준비된 로컬 세션을 만들고, session.register(...)는 이 세션에 Method를 추가합니다.
rank_method = python_method(
rank_values,
name="my_team.rank",
version="1.0.0",
inputs={
"data": DataInput(
columns={
"value": Column(source="measure"),
"group": Column(source="dimension"),
},
),
},
)
session = await practice_session()
session.register(rank_method)
print("등록된 Method:", [method.name for method in session.list_methods()])실행 결과
등록된 Method: ['my_team.rank']입력 선택과 실행
상품별 매출 순위를 구하도록 실제 데이터를 연결합니다.
| 설정 | 이 예제의 선택 |
|---|---|
bindings | value에 매출, group에 상품 |
params | descending=True: 매출이 큰 순서 |
scope | 조회 기간과 날짜 기준 |
Decision Layer는 선택한 데이터를 조회해 value, group 열이 있는 DataFrame을 만들고 rank_values에 전달합니다.
AI에게 “상품별 매출 순위를 알려줘”라고 요청할 때도 같은 흐름입니다. AI는 Method와 입력을 선택하고, 모호한 지표나 기간은 사용자에게 확인합니다. Decision Layer가 조회와 계산을 수행하면 AI는 결과를 읽고 설명합니다.
trial = await session.run(
"my_team.rank",
bindings={
"value": "cube://shop/sales/revenue",
"group": "cube://shop/sales/product",
},
params={"descending": True},
scope=Scope(
date_range=("2026-09-01", "2026-09-30"),
time_dimension="cube://shop/sales/date",
),
)
trial.result.primary.data실행 결과
[{'group': 'A', 'value': 30.0, 'rank': 1},
{'group': 'B', 'value': 20.0, 'rank': 2},
{'group': 'C', 'value': 20.0, 'rank': 2}]2. 여러 변수를 사용하는 OLS 분석
순위 계산은 연결 과정을 보여주기 위한 간단한 예제였습니다. 여러 변수와 분석 조건을 다루는 코드에서는 같은 방법을 재사용하는 일이 더 중요해집니다.
- 데이터 과학자가 검토한 코드를 한 번 등록합니다.
- 여러 사용자와 AI가 필요한 데이터를 연결해 같은 분석을 사용합니다.
- Decision Layer는 서버에서 실행한 입력과 결과를 Run에 기록합니다.
이번에는 고객별 광고 노출 횟수와 구매 금액의 관계를 OLS 회귀분석으로 살펴봅니다.
계산 함수
관심 변수와 결과의 관계를 구하면서 나이와 이전 구매 금액도 함께 고려합니다.
data: 분석할 DataFrameoutcome: 구매 금액 열treatment: 광고 노출 횟수 열confounders: 나이와 이전 구매 금액 열의 목록
함수는 열 이름을 인자로 받으므로 다른 데이터에도 재사용할 수 있습니다. 이 예제는 수치형 합성 데이터를 사용하며, 결과는 연관성을 나타냅니다. 인과관계를 입증하지는 않습니다.
def ols(data, outcome, treatment, confounders):
"""다른 변수들을 함께 고려해 관심 변수와 결과의 관계를 OLS로 분석합니다."""
import numpy as np
import pandas as pd
import statsmodels.api as sm
columns = [outcome, treatment, *confounders]
if len(set(columns)) != len(columns):
raise ValueError("같은 열을 두 번 선택했습니다.")
values = data[columns]
if not all(pd.api.types.is_numeric_dtype(values[col]) for col in columns):
raise ValueError("숫자형 열만 선택해 주세요.")
if not np.isfinite(values.to_numpy(dtype=float)).all():
raise ValueError("결측값이나 무한대를 먼저 처리해 주세요.")
if values[outcome].nunique() < 2:
raise ValueError("결과 변수의 값이 모두 같습니다.")
design = sm.add_constant(values[[treatment, *confounders]], has_constant="add")
if len(data) <= design.shape[1]:
raise ValueError(
"분석할 변수가 많습니다. 데이터를 늘리거나 변수를 줄여 주세요."
)
if np.linalg.matrix_rank(design) < design.shape[1]:
raise ValueError(
"설명변수들이 서로 완전히 겹칩니다. 변수 구성을 확인해 주세요."
)
fit = sm.OLS(values[outcome], design).fit()
return {
"value": float(fit.params[treatment]),
"n_observations": int(fit.nobs),
"r_squared": float(fit.rsquared),
"coefficients": {key: float(value) for key, value in fit.params.items()},
"causal_claim": "not_established",
"uncertainty": "not_reported",
}from examples.methods.authoring.customer_analysis import customer_sample
customers = customer_sample()
direct_ols = ols(
customers,
outcome="purchase_amount",
treatment="ad_exposures",
confounders=["age", "prior_purchase_amount"],
)
direct_ols실행 결과
{'value': 50.00000000000206,
'n_observations': 200,
'r_squared': 1.0,
'coefficients': {'const': 999.999999999618,
'ad_exposures': 50.00000000000206,
'age': 9.999999999999813,
'prior_purchase_amount': 0.20000000000000567},
'causal_claim': 'not_established',
'uncertainty': 'not_reported'}Method로 등록하기
entity_key는 고객별 한 행의 기준입니다. multiple=True인 confounders에는 여러 변수를 선택할 수 있습니다.
my_ols = python_method(
ols,
name="my_team.ols",
version="1.0.0",
inputs={
"data": DataInput(
columns={
"entity_id": Column(source="dimension", role="entity_key"),
"outcome": Column(source="measure", dtype="number"),
"treatment": Column(
source=["dimension", "measure"],
dtype="number",
),
"confounders": Column(
source=["dimension", "measure"],
dtype="number",
multiple=True,
),
},
minimum_rows=10,
)
},
output="estimate",
requires_packages={"statsmodels": ">=0.14,<0.16"},
)
session.register(my_ols)
print("등록된 Method:", [method.name for method in session.list_methods()])실행 결과
등록된 Method: ['my_team.rank', 'my_team.ols']입력 선택과 실행
고객 ID, 구매 금액과 광고 노출 횟수를 연결하고, 나이와 이전 구매 금액을 함께 선택합니다.
# 고객별 광고 노출 횟수와 구매 금액의 관계를 나이와 이전 구매 금액을 함께 고려해 분석합니다.
ols_trial = await session.run(
"my_team.ols",
bindings={
"entity_id": "cube://shop/customers/id",
"outcome": "cube://shop/customers/purchase_amount",
"treatment": "cube://shop/customers/ad_exposures",
"confounders": [
"cube://shop/customers/age",
"cube://shop/customers/prior_purchase_amount",
],
},
scope=Scope(
date_range=("2026-02-01", "2026-02-28"),
time_dimension="cube://shop/customers/date",
),
)
ols_trial.result.primary.data실행 결과
{'value': 50.00000000000206,
'n_observations': 200,
'r_squared': 1.0,
'coefficients': {'const': 999.999999999618,
'treatment': 50.00000000000206,
'confounders_0': 9.999999999999813,
'confounders_1': 0.20000000000000567},
'causal_claim': 'not_established',
'uncertainty': 'not_reported'}선택한 입력이 함수에 전달되기까지
1. 입력을 확인합니다.
bindings에 연결한 지표와 차원이 Cube에 정의되어 있고, Method의 입력 조건에 맞는지 확인합니다.
| 입력 | 연결한 데이터 |
|---|---|
entity_id | 고객 ID |
outcome | 구매 금액 |
treatment | 광고 노출 횟수 |
confounders | 나이, 이전 구매 금액 |
2. Cube에서 고객별 데이터를 조회합니다.
선택한 지표와 차원, 기간을 Cube 쿼리로 바꿉니다. Cube는 모델에 정의된 집계와 조인에 따라 고객별 데이터를 반환합니다.
Cube 조회 요청 보기
/cubejs-api/v1/load로 보내는 JSON입니다.
{
"query": {
"dimensions": ["customers.id", "customers.age"],
"measures": [
"customers.purchase_amount",
"customers.ad_exposures",
"customers.prior_purchase_amount"
],
"timeDimensions": [
{
"dimension": "customers.date",
"dateRange": ["2026-02-01", "2026-02-28"]
}
],
"order": [["customers.id", "asc"]],
"limit": 50000,
"offset": 0
}
}3. 열 이름을 바꿔 ols에 전달합니다.
구매 금액은 outcome, 광고 노출 횟수는 treatment가 됩니다. 나이와 이전 구매 금액은 선택한 순서대로 confounders_0, confounders_1에 담깁니다.
| entity_id | outcome | treatment | confounders_0 | confounders_1 |
|---|---|---|---|---|
| C001 | 8340 | 10 | 40 | 32200 |
| C002 | 10110 | 6 | 69 | 40600 |
Decision Layer가 이 표와 열 이름으로 함수를 호출합니다.
ols(
data=조회한_DataFrame,
outcome="outcome",
treatment="treatment",
confounders=["confounders_0", "confounders_1"],
)프로젝트에서 사용하기
예제에서는 로컬 세션에 두 Method를 등록했습니다. 관리자 화면과 AI에서도 사용하려면 함수와 등록 코드를 프로젝트의 method.py에 저장합니다.
프로젝트에 등록하기에서 파일 구성과 서버 실행 방법을 살펴봅니다.