【SQL周周练】给你无酸纸、变色油墨，你能伪造多少美金？

import numpy as np
import pandas as pd
import scipy
# 随机数种子
RANDOM_SEED = 2025
# 伪造开始日期
START_DATE = "2025-05-01"
# 伪造天数
NUM_DAY = 10
# 需要伪造的伪钞数量（张数，非金额）
NUM_TOTAL_COUNTERFEIT_CURRENCY = 1_000_000
# 一张伪钞需要多少无酸纸，简化问题只考虑重量(单位 g)
ACID_FREE_PAPER_EACH_COUNTERFEIT_CURRENCY = 1
# 所有伪钞需要的无酸纸(1.05 是一个冗余度，所有材料类似)
ACID_FREE_PAPER_ALL_NEED = (
    ACID_FREE_PAPER_EACH_COUNTERFEIT_CURRENCY 
    * NUM_TOTAL_COUNTERFEIT_CURRENCY 
    * 1.05
)
# 一张伪钞需要多少变色油墨，重量（单位 mg）
OPTICALLY_VARIABLE_INK_EACH_COUNTERFEIT_CURRENCY = 5
# 所有伪钞需要的变色油墨
OPTICALLY_VARIABLE_INK_ALL_NEED = (
    OPTICALLY_VARIABLE_INK_EACH_COUNTERFEIT_CURRENCY
    * NUM_TOTAL_COUNTERFEIT_CURRENCY
    * 1.05
)
# 一张伪钞需要多少安全线（单位 条）
SECURITY_THREAD_EACH_COUNTERFEIT_CURRENCY = 1
# 所有伪钞需要的防伪线
SECURITY_THREAD_ALL_NEED = (
    SECURITY_THREAD_EACH_COUNTERFEIT_CURRENCY 
    * NUM_TOTAL_COUNTERFEIT_CURRENCY 
    * 1.05
)

# 权重范围，用来随机生成数据（需要归一化）
WEIGHT_RANGE = (0.2, 2)
# 无酸纸每天供应的随机权重
acid_free_paper_supply_weight = scipy.stats.uniform.rvs(
    loc=WEIGHT_RANGE[0],
    scale=WEIGHT_RANGE[1] - WEIGHT_RANGE[0],
    size=NUM_DAY,
    random_state=RANDOM_SEED - 1,
)
# 变色油墨每天供应的权重
optically_variable_ink_supply_weight = scipy.stats.uniform.rvs(
    loc=WEIGHT_RANGE[0],
    scale=WEIGHT_RANGE[1] - WEIGHT_RANGE[0],
    size=NUM_DAY,
    random_state=RANDOM_SEED,
)
# 安全线每天供应的权重
security_thread_supply_weight = scipy.stats.uniform.rvs(
    loc=WEIGHT_RANGE[0],
    scale=WEIGHT_RANGE[1] - WEIGHT_RANGE[0],
    size=NUM_DAY,
    random_state=RANDOM_SEED + 1,
)
# 将权重归一化，使得所有天数的供应比例和为 1
acid_free_paper_supply_weight /= acid_free_paper_supply_weight.sum()
optically_variable_ink_supply_weight /= optically_variable_ink_supply_weight.sum()
security_thread_supply_weight /= security_thread_supply_weight.sum()

df = pd.DataFrame(
    {
        "acid_free_paper_supply": ACID_FREE_PAPER_ALL_NEED
        * acid_free_paper_supply_weight,
        "optically_variable_ink_supply": OPTICALLY_VARIABLE_INK_ALL_NEED
        * optically_variable_ink_supply_weight,
        "security_thread_supply": SECURITY_THREAD_ALL_NEED
        * security_thread_supply_weight
    }
)
# 四舍五入并转为 int
df = df.round().astype(int)
df["date"] = pd.date_range(start=START_DATE, periods=NUM_DAY, freq="D")
# 在 Jupyter 中展示数据
display(df)
out_csv_path = "./dwd_conterfeit_material_daily_supply_records.csv"
columns = [
    "date",
    "acid_free_paper_supply",
    "optically_variable_ink_supply",
    "security_thread_supply"
]
# 导出 csv 用来让 hive load 数据，utf-8-sig 编码处理中文，虽然表里数据没有中文
df[columns].to_csv(out_csv_path, header=False, index=False, encoding="utf-8-sig")

from pyhive import hive
# 配置连接参数
host_ip = "127.0.0.1"
port = 10000
username = "蒋点数分"
with hive.Connection(host=host_ip, port=port) as conn:
    cursor = conn.cursor()
    drop_table_sql = """
    drop table if exists data_exercise.dwd_conterfeit_material_daily_supply_records
    """
    print(drop_table_sql)
    cursor.execute(drop_table_sql)
    create_table_sql = """
    create table data_exercise.dwd_conterfeit_material_daily_supply_records (
        `date` string comment "日期",
        acid_free_paper_supply int comment "无酸纸供应量（单位g）",
        optically_variable_ink_supply int comment "变色油墨供应量（单位mg）",
        security_thread_supply int comment "安全线供应量"
    )
    comment "伪钞集团每天供应的伪钞原材料数量 | 文章编号：2c3d2561"
    row format delimited fields terminated by ","
    stored as textfile
    """
    
    print(create_table_sql)
    cursor.execute(create_table_sql)
    import os
    
    load_data_sql = f"""
    load data local inpath "{os.path.abspath(out_csv_path)}" 
    overwrite into table data_exercise.dwd_conterfeit_material_daily_supply_records
    """
    
    print(load_data_sql)
    cursor.execute(load_data_sql)
    cursor.close()

with calc_single_material_restrict_production as (
    -- 计算一种材料限制能造多少美元伪钞
    select
      `date`
    , acid_free_paper_supply
    , optically_variable_ink_supply
    , security_thread_supply
    -- 只考虑无酸纸，不考虑其他材料和每日最大制造量限制，累计伪钞制作数，下面以此类推
    -- 有些材料比例为 1，因此不额外写除以 1
    , sum(acid_free_paper_supply) over(orderby `date` asc) as cumulative_conterfeit_only_acid_free_paper
    -- 注意向下取整
    , floor(sum(optically_variable_ink_supply) over(orderby `date` asc) /5) as cumulative_conterfeit_only_optically_variable_ink
    , sum(security_thread_supply) over(orderby `date` asc) as cumulative_conterfeit_only_security_thread
    from data_exercise.dwd_conterfeit_material_daily_supply_records
)
, calc_all_restriction_prodection as (
    select
      `date`
    , acid_free_paper_supply
    , optically_variable_ink_supply
    , security_thread_supply
    , cumulative_conterfeit_only_acid_free_paper
    , cumulative_conterfeit_only_optically_variable_ink
    , cumulative_conterfeit_only_security_thread
    -- 使用 least 计算最小值
    , least(
        cumulative_conterfeit_only_acid_free_paper, 
        cumulative_conterfeit_only_optically_variable_ink, 
        cumulative_conterfeit_only_security_thread
      ) as cumulative_conterfeit_all_restriction
    from calc_single_material_restrict_production
)
select
  `date`
, cumulative_conterfeit_only_acid_free_paper
, cumulative_conterfeit_only_optically_variable_ink
, cumulative_conterfeit_only_security_thread
, cumulative_conterfeit_all_restriction
-- 减去上一行的数据，获取每日伪钞制造量
, cumulative_conterfeit_all_restriction -lag(cumulative_conterfeit_all_restriction, 1, 0) over(orderby `date` asc) as action_daily_production
, if( cumulative_conterfeit_all_restriction >=1000000, null, -- 已经完成目标量，就不写缺哪种材料了
    concat_ws(',', 
        if(cumulative_conterfeit_only_acid_free_paper=cumulative_conterfeit_all_restriction, '无酸纸', null),
        if(cumulative_conterfeit_only_optically_variable_ink=cumulative_conterfeit_all_restriction, '变色油墨', null),
        if(cumulative_conterfeit_only_security_thread=cumulative_conterfeit_all_restriction, '安全线', null)
    )
) as `最缺的材料`
from calc_all_restriction_prodection

with hive.Connection(host=host_ip, port=port) as conn:
    select_data_sql = ''' 我给出 SQL 答案 '''
    df_outcome = pd.read_sql_query(select_data_sql, conn)
from pyvchart import render_chart
spec = {
"type": 'area',
"data": [
    {
      "id": 'lineData',
      "values": pd.melt(df_outcome[[
        'date','cumulative_conterfeit_only_acid_free_paper',
        'cumulative_conterfeit_only_optically_variable_ink',
        'cumulative_conterfeit_only_security_thread'
      ]], id_vars=['date']).to_dict(orient='records')
    },
    {
      "id": 'areaData',
      "values": pd.melt(df_outcome[['date','cumulative_conterfeit_all_restriction',
        '最缺的材料']], id_vars=['date']).to_dict(orient='records')
    },
  ],
"series": [
    {
      "type": 'line',
      "dataId": 'lineData',
      "xField": 'date',
      "yField": 'value',
      "seriesField": 'variable',
    },
    {
      "type": 'area',
      "dataId": 'areaData',
      "xField": 'date',
      "yField": 'value',
      "seriesField": 'variable',
    },
 ],
};
display(render_chart(spec))