🤖️ Qbot - AI 自动量化交易机器人

from dagster import asset
from pandas import DataFrame, read_html,
                  get_dummies
from sklearn.linear_model import LinearRegression as Regression
                

@asset
def country_stats() -> DataFrame:
 df = read_html("https://tinyurl.com/mry64ebh")[0]
 df.columns = ["country", "continent", "pop_change"]
 df["pop_change"] = df["pop_change"].str.rstrip("%").astype("float")
 return df

@asset
def change_model(country_stats: DataFrame) -> Regression:
 data = country_stats.dropna(subset=["pop_change"])
 dummies = get_dummies(data[["continent"]])
 return Regression().fit(dummies, data["pop_change"])

@asset
def continent_stats(country_stats: DataFrame, change_model: Regression) -> DataFrame:
 result = country_stats.groupby("continent").sum()
 result["pop_change_factor"]
                  = change_model.coef_
 return result
                

Materialize All

Materializing an asset launches a run and saves the results to persistent storage. Trigger materializations right from any asset graph.

from dagster_airbyte import build_airbyte_assets
from dagster_dbt import load_assets_from_dbt_project
from dagster import asset, Definitions
from .constants import AIRBYTE_CONNECTION_ID, DBT_PROJECT_DIR

airbyte_assets = build_airbyte_assets(
 connection_id=AIRBYTE_CONNECTION_ID,
 destination_tables=["orders", "users"],
 asset_key_prefix=["postgres_replica"],
)

dbt_assets = load_assets_from_dbt_project(project_dir=DBT_PROJECT_DIR)

@asset(compute_kind="python")
def order_forecast_model(daily_order_summary: pd.DataFrame) -> np.ndarray:
 train_set = daily_order_summary.to_numpy()
                
 xdata, ydata = train_set[:, 0], train_set[:, 2]
 return optimize.curve_fit(f=model_func, xdata=xdata, ydata=ydata, p0=[10, 100])[0]

defs = Definitions(
 assets=[*airbyte_assets, *dbt_assets, order_forecast_model]
)

Materialize All

Materializing an asset launches a run and saves the results to persistent storage. Trigger materializations right from any asset graph.

from dagster import job, op
from dagster_snowflake import snowflake_resource
from dagster_slack import slack_resource
from .config import SLACK_CONFIG,
                  SNOWFLAKE_CONFIG


@op(required_resource_keys={"snowflake"})
def find_stale_tables(context) -> list[str]:
 return [
 record[0]
 for record in context.resources.snowflake.execute_query(
 """select table_schema || '.' || table_name
 from information_schema.tables
 where last_altered > dateadd(day, -30,
                    current_timestamp)""",
 fetch_results=True,
 )
 ]


@op(required_resource_keys={"slack"})
def post_stale_tables(context, tables: List[str]) -> None:
 context.resources.slack.chat_postMessage(
 channel="#stale-tables", text="Stale tables:\n" + "\n- ".join(tables)
 )


@job(
 resource_defs={
 "snowflake": snowflake_resource.configured(SNOWFLAKE_CONFIG),
 "slack": slack_resource.configured(SLACK_CONFIG),
                
 }
)
def report_stale_tables():
 post_stale_tables(find_stale_tables())