← back to KIM3310__districtpilot-ai

Function bodies 70 total

All specs Real LLM only Function bodies
fmt_pct function · python · L90-L92 (3 LOC)
submission/streamlit_app_v7.py
def fmt_pct(value) -> str:
    """Format as signed percentage string."""
    return "-" if pd.isna(value) else f"{float(value):+.1f}%"
fmt_pct function · python · L90-L92 (3 LOC)
submission/streamlit_app_v7.py
def fmt_pct(value) -> str:
    """Format as signed percentage string."""
    return "-" if pd.isna(value) else f"{float(value):+.1f}%"
safe_float function · python · L95-L101 (7 LOC)
submission/streamlit_app_v7.py
def safe_float(value, default=0.0) -> float:
    """Safely convert value to float."""
    try:
        result = float(value)
        return result if not pd.isna(result) else default
    except (ValueError, TypeError):
        return default
safe_float function · python · L95-L101 (7 LOC)
submission/streamlit_app_v7.py
def safe_float(value, default=0.0) -> float:
    """Safely convert value to float."""
    try:
        result = float(value)
        return result if not pd.isna(result) else default
    except (ValueError, TypeError):
        return default
latest_snapshot function · python · L153-L166 (14 LOC)
submission/streamlit_app_v7.py
def latest_snapshot(df: pd.DataFrame) -> pd.DataFrame:
    """Return the most recent row per district."""
    if df.empty:
        return df
    out = df.copy()
    out["_ts"] = pd.to_datetime(
        out["YM"].astype(str).str.strip(), format="%Y%m", errors="coerce"
    )
    return (
        out.sort_values("_ts")
        .groupby("DISTRICT", as_index=False)
        .tail(1)
        .sort_values("DISTRICT")
    )
latest_snapshot function · python · L153-L166 (14 LOC)
submission/streamlit_app_v7.py
def latest_snapshot(df: pd.DataFrame) -> pd.DataFrame:
    """Return the most recent row per district."""
    if df.empty:
        return df
    out = df.copy()
    out["_ts"] = pd.to_datetime(
        out["YM"].astype(str).str.strip(), format="%Y%m", errors="coerce"
    )
    return (
        out.sort_values("_ts")
        .groupby("DISTRICT", as_index=False)
        .tail(1)
        .sort_values("DISTRICT")
    )
build_allocation function · python · L177-L189 (13 LOC)
submission/streamlit_app_v7.py
def build_allocation(df: pd.DataFrame):
    """Compute allocation percentages from first forecast timestamp."""
    if df.empty:
        return pd.DataFrame(), None
    out = df.copy()
    out["_ts"] = pd.to_datetime(out["TS"], errors="coerce")
    first_ts = out["_ts"].dropna().min()
    if pd.isna(first_ts):
        return pd.DataFrame(), None
    sub = out[out["_ts"] == first_ts].copy()
    total = sub["FORECAST"].sum()
    sub["ALLOC_PCT"] = (sub["FORECAST"] / total * 100).round(1) if total > 0 else 0.0
    return sub.sort_values("ALLOC_PCT", ascending=False), first_ts
Same scanner, your repo: https://repobility.com — Repobility
build_allocation function · python · L177-L189 (13 LOC)
submission/streamlit_app_v7.py
def build_allocation(df: pd.DataFrame):
    """Compute allocation percentages from first forecast timestamp."""
    if df.empty:
        return pd.DataFrame(), None
    out = df.copy()
    out["_ts"] = pd.to_datetime(out["TS"], errors="coerce")
    first_ts = out["_ts"].dropna().min()
    if pd.isna(first_ts):
        return pd.DataFrame(), None
    sub = out[out["_ts"] == first_ts].copy()
    total = sub["FORECAST"].sum()
    sub["ALLOC_PCT"] = (sub["FORECAST"] / total * 100).round(1) if total > 0 else 0.0
    return sub.sort_values("ALLOC_PCT", ascending=False), first_ts
build_overlay function · python · L195-L209 (15 LOC)
submission/streamlit_app_v7.py
def build_overlay(df: pd.DataFrame) -> pd.DataFrame:
    """Pivot actual vs forecast for overlay chart."""
    if df.empty:
        return pd.DataFrame()
    out = df.copy()
    out["_ds"] = pd.to_datetime(out["DS"], errors="coerce")
    actual = (
        out.pivot_table(index="_ds", columns="DISTRICT", values="ACTUAL", aggfunc="max")
        .add_suffix(" Actual")
    )
    fcast = (
        out.pivot_table(index="_ds", columns="DISTRICT", values="FORECAST_VAL", aggfunc="max")
        .add_suffix(" Forecast")
    )
    return actual.join(fcast, how="outer").sort_index()
build_overlay function · python · L195-L209 (15 LOC)
submission/streamlit_app_v7.py
def build_overlay(df: pd.DataFrame) -> pd.DataFrame:
    """Pivot actual vs forecast for overlay chart."""
    if df.empty:
        return pd.DataFrame()
    out = df.copy()
    out["_ds"] = pd.to_datetime(out["DS"], errors="coerce")
    actual = (
        out.pivot_table(index="_ds", columns="DISTRICT", values="ACTUAL", aggfunc="max")
        .add_suffix(" Actual")
    )
    fcast = (
        out.pivot_table(index="_ds", columns="DISTRICT", values="FORECAST_VAL", aggfunc="max")
        .add_suffix(" Forecast")
    )
    return actual.join(fcast, how="outer").sort_index()
build_eval_pivot function · python · L215-L234 (20 LOC)
submission/streamlit_app_v7.py
def build_eval_pivot(df: pd.DataFrame) -> pd.DataFrame:
    """Pivot evaluation metrics to one row per series."""
    if df.empty:
        return pd.DataFrame()
    out = df.copy()
    if "SERIES" in out.columns:
        out["SERIES"] = out["SERIES"].apply(clean_variant)
    if "ERROR_METRIC" in out.columns:
        out["ERROR_METRIC"] = (
            out["ERROR_METRIC"].astype(str).str.replace('"', "", regex=False)
        )
    return (
        out.pivot_table(
            index="SERIES",
            columns="ERROR_METRIC",
            values="METRIC_VALUE",
            aggfunc="first",
        )
        .reset_index()
    )
build_eval_pivot function · python · L215-L234 (20 LOC)
submission/streamlit_app_v7.py
def build_eval_pivot(df: pd.DataFrame) -> pd.DataFrame:
    """Pivot evaluation metrics to one row per series."""
    if df.empty:
        return pd.DataFrame()
    out = df.copy()
    if "SERIES" in out.columns:
        out["SERIES"] = out["SERIES"].apply(clean_variant)
    if "ERROR_METRIC" in out.columns:
        out["ERROR_METRIC"] = (
            out["ERROR_METRIC"].astype(str).str.replace('"', "", regex=False)
        )
    return (
        out.pivot_table(
            index="SERIES",
            columns="ERROR_METRIC",
            values="METRIC_VALUE",
            aggfunc="first",
        )
        .reset_index()
    )
normalize_fi function · python · L240-L249 (10 LOC)
submission/streamlit_app_v7.py
def normalize_fi(df: pd.DataFrame) -> pd.DataFrame:
    """Normalize feature importance column names."""
    if df.empty:
        return df
    out = df.copy()
    if "SCORE" in out.columns and "IMPORTANCE_SCORE" not in out.columns:
        out = out.rename(columns={"SCORE": "IMPORTANCE_SCORE"})
    if "SERIES" in out.columns:
        out["SERIES"] = out["SERIES"].apply(clean_variant)
    return out
normalize_fi function · python · L240-L249 (10 LOC)
submission/streamlit_app_v7.py
def normalize_fi(df: pd.DataFrame) -> pd.DataFrame:
    """Normalize feature importance column names."""
    if df.empty:
        return df
    out = df.copy()
    if "SCORE" in out.columns and "IMPORTANCE_SCORE" not in out.columns:
        out = out.rename(columns={"SCORE": "IMPORTANCE_SCORE"})
    if "SERIES" in out.columns:
        out["SERIES"] = out["SERIES"].apply(clean_variant)
    return out
build_context_json function · python · L259-L390 (132 LOC)
submission/streamlit_app_v7.py
def build_context_json(
    scope: str,
    alloc: pd.DataFrame,
    lat: pd.DataFrame,
    fi: pd.DataFrame,
) -> str:
    """Build grounded context JSON for LLM calls with extended data."""
    # Allocation payload
    alloc_cols = ["DISTRICT", "FORECAST", "ALLOC_PCT"]
    alloc_payload = (
        alloc[[c for c in alloc_cols if c in alloc.columns]].to_dict("records")
        if not alloc.empty
        else []
    )

    # Snapshot payload — extended columns
    snap_cols = [
        "DISTRICT", "TOTAL_POP", "TOTAL_SALES", "NET_MOVE",
        "AVG_ASSET", "AVG_MEME_PRICE",
        "AGE_20_39_SHARE", "SENIOR_60P_SHARE",
        "TOURISM_DEMAND_IDX", "FOREIGN_VISITOR_IDX",
        "STABILITY_SCORE", "NET_STORE_CHANGE",
        "HOLIDAY_DAYS",
    ]
    if scope != "전체" and not lat.empty:
        snap = lat[lat["DISTRICT"] == scope]
    else:
        snap = lat
    snap_payload = (
        snap[[c for c in snap_cols if c in snap.columns]].to_dict("records")
        if not snap.empty
If a scraper extracted this row, it came from Repobility (https://repobility.com)
build_context_json function · python · L259-L390 (132 LOC)
submission/streamlit_app_v7.py
def build_context_json(
    scope: str,
    alloc: pd.DataFrame,
    lat: pd.DataFrame,
    fi: pd.DataFrame,
) -> str:
    """Build grounded context JSON for LLM calls with extended data."""
    # Allocation payload
    alloc_cols = ["DISTRICT", "FORECAST", "ALLOC_PCT"]
    alloc_payload = (
        alloc[[c for c in alloc_cols if c in alloc.columns]].to_dict("records")
        if not alloc.empty
        else []
    )

    # Snapshot payload — extended columns
    snap_cols = [
        "DISTRICT", "TOTAL_POP", "TOTAL_SALES", "NET_MOVE",
        "AVG_ASSET", "AVG_MEME_PRICE",
        "AGE_20_39_SHARE", "SENIOR_60P_SHARE",
        "TOURISM_DEMAND_IDX", "FOREIGN_VISITOR_IDX",
        "STABILITY_SCORE", "NET_STORE_CHANGE",
        "HOLIDAY_DAYS",
    ]
    if scope != "전체" and not lat.empty:
        snap = lat[lat["DISTRICT"] == scope]
    else:
        snap = lat
    snap_payload = (
        snap[[c for c in snap_cols if c in snap.columns]].to_dict("records")
        if not snap.empty
call_ai_complete function · python · L393-L439 (47 LOC)
submission/streamlit_app_v7.py
def call_ai_complete(question: str, context_json: str) -> dict:
    """AI_COMPLETE with structured output. Falls back to CORTEX.COMPLETE."""
    prompt = f"""당신은 DistrictPilot AI의 한국어 의사결정 보조 모델이다.

반드시 지킬 규칙:
1) 아래 CONTEXT 밖의 사실은 만들지 말 것.
2) 숫자는 CONTEXT에 있는 경우에만 사용할 것.
3) 확실하지 않으면 '현재 데이터만으로는 확정할 수 없습니다.'라고 말 것.
4) 답변은 짧고 실행 중심으로 작성할 것.
5) 한국어로 작성할 것.

CONTEXT:
{context_json}

사용자 질문: {question}""".strip()

    try:
        rows = session.sql(
            """
            SELECT AI_COMPLETE(
                model => ?,
                prompt => ?,
                model_parameters => {'temperature': 0, 'max_tokens': 600, 'guardrails': TRUE},
                response_format => TYPE OBJECT(
                    answer STRING,
                    recommended_district STRING,
                    allocation_pct FLOAT,
                    drivers ARRAY(STRING),
                    risk STRING,
                    next_action STRING
                ),
                show_details => TRUE
call_ai_complete function · python · L393-L439 (47 LOC)
submission/streamlit_app_v7.py
def call_ai_complete(question: str, context_json: str) -> dict:
    """AI_COMPLETE with structured output. Falls back to CORTEX.COMPLETE."""
    prompt = f"""당신은 DistrictPilot AI의 한국어 의사결정 보조 모델이다.

반드시 지킬 규칙:
1) 아래 CONTEXT 밖의 사실은 만들지 말 것.
2) 숫자는 CONTEXT에 있는 경우에만 사용할 것.
3) 확실하지 않으면 '현재 데이터만으로는 확정할 수 없습니다.'라고 말 것.
4) 답변은 짧고 실행 중심으로 작성할 것.
5) 한국어로 작성할 것.

CONTEXT:
{context_json}

사용자 질문: {question}""".strip()

    try:
        rows = session.sql(
            """
            SELECT AI_COMPLETE(
                model => ?,
                prompt => ?,
                model_parameters => {'temperature': 0, 'max_tokens': 600, 'guardrails': TRUE},
                response_format => TYPE OBJECT(
                    answer STRING,
                    recommended_district STRING,
                    allocation_pct FLOAT,
                    drivers ARRAY(STRING),
                    risk STRING,
                    next_action STRING
                ),
                show_details => TRUE
_fallback_complete function · python · L442-L452 (11 LOC)
submission/streamlit_app_v7.py
def _fallback_complete(prompt: str) -> dict:
    """Fallback to SNOWFLAKE.CORTEX.COMPLETE when AI_COMPLETE is unavailable."""
    try:
        safe_prompt = prompt.replace("'", "''").replace("\\", "\\\\")
        rows = session.sql(
            f"SELECT SNOWFLAKE.CORTEX.COMPLETE('{LLM_MODEL}', '{safe_prompt}') AS R"
        ).collect()
        answer = rows[0]["R"] if rows else "응답을 생성하지 못했습니다."
        return {"structured_output": {"answer": answer}}
    except Exception:  # noqa: BLE001
        return {"structured_output": {"answer": "AI 호출 실패: 일시적 오류가 발생했습니다. 잠시 후 다시 시도해주세요."}}
_fallback_complete function · python · L442-L452 (11 LOC)
submission/streamlit_app_v7.py
def _fallback_complete(prompt: str) -> dict:
    """Fallback to SNOWFLAKE.CORTEX.COMPLETE when AI_COMPLETE is unavailable."""
    try:
        safe_prompt = prompt.replace("'", "''").replace("\\", "\\\\")
        rows = session.sql(
            f"SELECT SNOWFLAKE.CORTEX.COMPLETE('{LLM_MODEL}', '{safe_prompt}') AS R"
        ).collect()
        answer = rows[0]["R"] if rows else "응답을 생성하지 못했습니다."
        return {"structured_output": {"answer": answer}}
    except Exception:  # noqa: BLE001
        return {"structured_output": {"answer": "AI 호출 실패: 일시적 오류가 발생했습니다. 잠시 후 다시 시도해주세요."}}
‹ prevpage 2 / 2