来源：和讯股票

近期，中国证监会北京监管局发布了一则行政处罚决定书，开源证券年内受罚清单上再添一笔。决定书中显示，开源证券固定收益部部门经理蒋东波在职期间，曾于2016年10月25日至2017年2月20日利用其母证券帐户违法持有、买卖股票14只，获利64.86万元。上述行为违反了《证券法》规定，构成 “法律、行政法规规定禁止参与股票交易的人员，直接或者以化名、借他人名义持有、买卖股票”的行为。并且在调查中意图隐瞒、掩盖违法事实，干扰调查工作。

因此，北京证监局在经过调查后依据《证券法》相关规定，对蒋东波作出行政处罚，同时没收违法所得64.86万元，并处以罚金259.45万元。几乎达到了一至五倍的罚款最高额。

公司规模中下 前三年营收净利连续下滑

据开源证券官网显示，开源证券股份有限公司成立于1994年2月，注册资本金27.55亿，公司总部位于陕西省西安市高新区，目前下设78家分支机构，以及4家控、参股子公司。公司大股东为陕西煤业化工集团有限责任公司，持股比例51%，第二大股东为佛山市顺德区金盛瑞泰投资有限公司，持股比例35.35%。

现任董事长兼总经理为李刚。2007年1月进入开源证券，先后担任增资扩股负责人、总支书记、董事、董事长特别助理、执行委员会主任委员，2010年6月至今,任开源证券董事长，至今已有九年。

同时通过中国证券业协会公开数据可以得知，开源证券行业排名位于中下游，2019年之前三年净利连续下滑。中国证券业协会发布的2018年度证券公司经营业绩排名情况显示，98家证券公司中，开源证券总资产1,643,702 万元，排名第68位；净资产515,562万元，排名第69名；2018年度其实现营收131,275万元，排名第53位；实现净利润4,017万元，排名第66位；净资产收益率为0.94%，排名第64位；净资本409,502万元，排名第72位；核心净资本排名第70位，为409,502万元。在此之前的2017年，开源证券实现营业收入近10亿，同比下降7.09%，净利润0.56亿，同比下降69.78%；2016年，开源证券实现实现营业收入10.8亿，同比增长20.61%，净利润1.8亿，同比下降28.02%。

业绩猛增 合规却成后遗症

2019年上半年，开源证券以“大投行、大资管”定位自身，依托上半年行业利好不断，在传统经纪业务佣金率不断下滑的大背景下，摆脱了单一的经纪业务发展模式，创造了业内罕见的“开源速度”。实现营业收入10.7亿元，同比增长175%；净利润3.8亿元，同比增长900%。然而在超速发展之下，合规风控却似乎成了开源证券最大的后遗症。

据公开资料显示，算上本次员工炒股问题，开源证券2019年在证监系统处罚上已是“五进宫”，同时距上次被罚仅一个月。

今年4月，开源证券在资管业务、债券交易业务、公司内控等方面存在8项违规，被陕西证监局要求整改；8月，因债券承销与受托管理、资产证券化业务多项违约，开源证券再一次被陕西证监局处罚；9月，开源证券作为16洪业债的承销机构，未审慎调查洪业化工有关关联方关系及对外担保等事项被山东证监局处罚；11月，开源证券及分管副总杨彬、项目负责人杭宇在承销发行人天津市浩通物产有限公司的三期公司债券中，尽职调查未勤勉尽责同时未及时发布临时受托管理事务报告被天津证监局责令改正。

证券从业人员违规炒股屡禁不止 争议不断

事实上，证券从业人员炒股一直都处于监管红线。据统计，今年以来，仅证监会就发出了6张针对证券从业人员违规炒股的罚单，罚款金额最高达千万，另外各地证监局也有相关罚单。

其实，证券从业人员炒股是否应该一刀切禁止，业界仍存在一定争议。一方面，证券从业人员拥有一定信息优势，可能利用内幕信息进行交易；另一方面，证券从业人员也有一定正当的个人投资需求，并且并非所有证券从业人员都能接触到内幕信息。

某证券法律合规部员工曾发文表示，目前行业对证券从业人员个人投资行为的管理措施，主要围绕落实《证券法》第四十三条禁止股票投资的规定展开。该条的立法目的在于管理证券从业人员执业过程中面临的利益冲突。禁止性的法律规范，优点在于设立了简单明了的行动指引，但随着证券市场的日益复杂，这样的立法往往容易脱离预设目标，增加监管成本。他建议，明确敏感信息的界定标准，建立多层次的敏感信息管控体系。对证券从业人员个人投资行为进行管理的目的是为了防范从业人员利用其信息优势侵占公司、客户及其他社会公众的利益。通过“敏感信息”管理，平衡从业人员个人与所属公司、客户之间的利益冲突，才是证券从业人员个人投资行为管理的核心。

前两期中，我们为大家展示了如何用 Window Join、Asof Join 引擎将逐笔成交数据与快照数据进行关联分析，以及使用 Left Semi Join 补充原始委托信息，这些都是金融中常见的应用场景，感兴趣的小伙伴可以阅读往期内容。

本期，我们将为大家介绍剩下的三个场景：

- 实时计算股票与某指数的分钟收益率相关性；

- 对多个数据源降频采样，计算分钟指标并将结果关联到同一张表中；

- 根据快照数据实时匹配股票历史日频指标。

在量化实盘中，往往需要对原始的行情快照、逐笔成交数据进行降采样处理，以得到分钟频指标，作为策略研发的输入数据，这就要求将多个不同数据源计算出的指标关联到同一张表中。

本例将对快照和成交数据分别做实时的 1 分钟聚合，并将计算所得指标关联后输出到同一张宽表中。

这个场景的特征是，每支股票的快照和逐笔分钟指标在每一分钟只有一条记录，具有唯一性，并且在某一分钟的输出上，期望总是在两类指标都计算完成后再关联输出。

用 Equal Join 引擎实现此场景的脚本如下：

// create table share streamTable(1:0, `Sym`TradeTime`Side`TradeQty, [SYMBOL, TIME, INT, LONG]) as trades share streamTable(1:0, `UpdateTime`Sym`BuyTradeQty`SellTradeQty, [TIME, SYMBOL, LONG, LONG]) as tradesMin share streamTable(1:0, `Sym`Time`Bid1Price`Bid1Qty, [SYMBOL, TIME, DOUBLE, LONG]) as snapshot share streamTable(1:0, `UpdateTime`Sym`AvgBid1Amt, [TIME, SYMBOL, DOUBLE]) as snapshotMin share streamTable(1:0, `UpdateTime`Sym`AvgBid1Amt`BuyTradeQty`SellTradeQty, [TIME, SYMBOL, DOUBLE, LONG, LONG]) as output // create engine: eqJoinEngine = createEqualJoinEngine(name="EqualJoin", leftTable=tradesMin, rightTable=snapshotMin, outputTable=output, metrics=<[AvgBid1Amt, BuyTradeQty, SellTradeQty]>, matchingColumn=`Sym, timeColumn=`UpdateTime) // create engine: tsEngine1 = createTimeSeriesEngine(name="tradesAggr", windowSize=60000, step=60000, metrics=<[sum(iif(Side==1, 0, TradeQty)), sum(iif(Side==2, 0, TradeQty))]>, dummyTable=trades, outputTable=getLeftStream(eqJoinEngine), timeColumn=`TradeTime, keyColumn=`Sym, useSystemTime=false, fill=(0, 0)) // create engine: tsEngine2 = createTimeSeriesEngine(name="snapshotAggr", windowSize=60000, step=60000, metrics=<[avg(iif(Bid1Price!=NULL, Bid1Price*Bid1Qty, 0))]>, dummyTable=snapshot, outputTable=getRightStream(eqJoinEngine), timeColumn=`Time, keyColumn=`Sym, useSystemTime=false, fill=(0.0)) // subscribe topic subscribeTable(tableName="trades", actionName="minAggr", handler=tsEngine1, msgAsTable=true, offset=-1, hash=1) subscribeTable(tableName="snapshot", actionName="minAggr", handler=tsEngine2, msgAsTable=true, offset=-1, hash=2)

首先用两个独立的时序聚合引擎（createTimeSeriesEngine）对原始的快照和成交数据流按数据中的时间戳做实时聚合，输出每一分钟的指标；然后通过引擎级联的方式，将两个时序聚合引擎的输出分别作为左右表注入连接引擎。

构造数据写入作为原始输入的 2 个流数据表，先写入右表，再写入左表：

// generate data: snapshot t1 = table(`A`B`A`B`A`B as Sym, 10:00:52.000+(3 3 6 6 9 9)*1000 as Time, (3.5 7.6 3.6 7.6 3.6 7.6) as Bid1Price, (1000 2000 500 1500 400 1800) as Bid1Qty) // generate data: trade t2 = table(`A`A`B`A`B`B`A`B`B`A as Sym, 10:00:54.000+(1..10)*700 as TradeTime, (1 2 1 1 1 1 2 1 2 2) as Side, (1..10) * 10 as TradeQty) // input trades.append!(t2) snapshot.append!(t1)

关联得到的结果表 output 如下：

在当日的实时计算中，有时还需要依赖历史指标，本例中我们基于行情快照数据，通过匹配股票代码关联昨日的日频指标。

这个场景的特征是，每条快照记录到达后要求立刻关联输出，如果日频数据里没有对应的股票，输出结果对应的字段为空，输出与原始输入中的每一条行情快照记录一一对应。同时，日频指标并非实时数据，而是一个以较低频率更新的有主键的离线数据集。

用 Lookup Join 引擎实现此场景的脚本如下：

// create table share streamTable(1:0, `Sym`Time`Open`High`Low`Close, [SYMBOL, TIME, DOUBLE, DOUBLE, DOUBLE, DOUBLE]) as snapshot historicalData = table(`A`B as Sym, (0.8 0.2) as PreWeight, (3.1 7.6) as PreClose) share table(1:0, `Sym`Time`Open`High`Low`Close`PreWeight`PreClose, [SYMBOL, TIME, DOUBLE, DOUBLE, DOUBLE, DOUBLE, DOUBLE, DOUBLE]) as output // create engine lookupJoinEngine = createLookupJoinEngine(name="lookupJoin", leftTable=snapshot, rightTable=historicalData, outputTable=output, metrics=<[Time, Open, High, Low, Close, PreWeight, PreClose]>, matchingColumn=`Sym, checkTimes=10s) // subscribe topic subscribeTable(tableName="snapshot", actionName="appendLeftStream", handler=getLeftStream(lookupJoinEngine), msgAsTable=true, offset=-1)

构造数据写入作为引擎左表输入的流数据表 snapshot：

// generate data: snapshot t1 = table(`A`B`A`B`A`B as Sym, 10:00:00.000+(3 3 6 6 9 9)*1000 as Time, (3.5 7.6 3.5 7.6 3.5 7.6) as Open, (3.5 7.6 3.6 7.6 3.6 7.6) as High, (3.5 7.6 3.5 7.6 3.4 7.5) as Low, (3.5 7.6 3.5 7.6 3.6 7.5) as Close) snapshot.append!(t1)

输入数据与关联关系如下：

结果在左表数据到达引擎时立刻输出，关联得到的结果表 output 如下：

Left Semi Join 引擎的连接机制类似于 SQL 中的 equal join ，按连接列等值关联左右表。上一期我们利用这一功能实现了在成交数据的基础上匹配委托订单，丰富原始的委托信息。本例中，我们来实时计算股票和某个指数在过去一段时间内分钟收益率的相关性。

这个场景的特征是，两个数据流的时间戳频率一致，全部股票都需要关联同一支指数，输入是已经降为分钟频率的股票和指数数据，输出与原始输入中的股票数据一一对应。

可以用如下脚本实现关联：

// create table share streamTable(1:0, `Sym`Time`Close, [SYMBOL, TIME, DOUBLE]) as stockKline share streamTable(1:0, `Sym`Time`Close, [SYMBOL, TIME, DOUBLE]) as indexKline share streamTable(1:0, `Time`Sym`Close`Index1Close, [TIME, SYMBOL, DOUBLE, DOUBLE]) as stockKlineAddIndex1 share streamTable(1:0, `Sym`Time`Close`Index1Close`Index1Corr, [SYMBOL, TIME, DOUBLE, DOUBLE, DOUBLE]) as output // create engine: calculate correlation rsEngine = createReactiveStateEngine(name="calCorr", dummyTable=stockKlineAddIndex1, outputTable=output, metrics=[<Time>, <Close>, <Index1Close>, <mcorr(ratios(Close)-1, ratios(Index1Close)-1, 3)>], keyColumn="Sym") // create engine: left join Index1 ljEngine1 = createLeftSemiJoinEngine(name="leftJoinIndex1", leftTable=stockKline, rightTable=indexKline, outputTable=getStreamEngine("calCorr"), metrics=<[Sym, Close, indexKline.Close]>, matchingColumn=`Time) // subscribe topic def appendIndex(engineName, indexName, msg){ tmp = select * from msg where Sym = indexName getRightStream(getStreamEngine(engineName)).append!(tmp) } subscribeTable(tableName="indexKline", actionName="appendIndex1", handler=appendIndex{"leftJoinIndex1", "idx1"}, msgAsTable=true, offset=-1, hash=1) subscribeTable(tableName="stockKline", actionName="appendStock", handler=getLeftStream(ljEngine1), msgAsTable=true, offset=-1, hash=0)

这里连接引擎的输出会直接注入响应式状态引擎进行下一步计算，多个引擎之间采用了引擎级联的方式处理。

构造数据写入作为原始输入的 2 个流数据表：

// generate data: stock Kline t1 = table(`A`B`A`B`A`B`A`B`A`B as Sym, 10:00:00.000+(0 0 1 1 2 2 3 3 4 4)*60000 as Time, (4.1 7.6 3.8 7.6 4.3 7.5 3.5 7.6 4.2 7.6) as Close) // generate data: index Kline t2 = table(`idx1`idx2`idx1`idx2`idx1`idx2`idx1`idx2`idx1`idx2 as Sym, 10:00:00.000+(0 0 1 1 2 2 3 3 4 4)*60000 as Time, (2.1 5 2.2 5 1.9 5 1.7 5 1.7 5) as Close) // input data indexKline.append!(t2) stockKline.append!(t1)

输入数据与关联关系如下：

关联得到的结果表 output 如下：

至此，我们已经为大家展示了所有六个应用场景，囊括了 Asof Join、Window Join、Equal Join、Lookup Join、Left Semi Join 这五个流数据连接引擎。

这些引擎均内置实现了高效的关联计算、实时触发规则和内存管理机制，很好地解决了对齐难、触发难、缓存难、计算难等一系列问题。开发人员通过简单的引擎参数配置，便能快速实现复杂的实时关联需求。

在这些连接引擎的基础上，再结合 DolphinDB 流数据框架中其他流计算引擎、流水线处理、并行计算等重要特性，开发人员便可以高效实现业务场景实时化，掌握更及时的信息、挖掘更多的业务价值。

五大连接引擎的信息对比如下表所示：

上一篇：手机怎么炒股开户和怎么炒股开户

下一篇：什么是炒股打板 炒股用什么软件